| """Automatic application and reversible exclusion of collected return evidence.""" |
| import json |
| from datetime import datetime |
|
|
| from sqlalchemy import update |
|
|
| from models import ReturnDocument, ReturnAudit |
| from shareholder_returns import load_documents |
|
|
|
|
| def automatic_status(document): |
| if document.get("error"): |
| return "reference" |
| if document.get("is_withdrawal"): |
| return "active" if document.get("supersedes_id") else "reference" |
| events = document.get("events") or [] |
| policies = document.get("policies") or [] |
| usable_event = any(e.get("scope") == "common" and e.get("amount") is not None for e in events) |
| usable_policy = any(p.get("scope") == "common" and (p.get("amount") is not None or p.get("ratio") is not None or p.get("kind") == "none") for p in policies) |
| return "active" if usable_event or usable_policy else "reference" |
|
|
|
|
| def apply_pending_documents(db): |
| """Upgrade existing candidates once, in bulk; never override a user's exclusion.""" |
| documents = [d for d in load_documents(db) if d["status"] == "pending"] |
| now = datetime.utcnow() |
| groups = {"active": [], "reference": []} |
| for before in documents: |
| status = automatic_status(before) |
| groups[status].append(before["id"]) |
| after = {**before, "status": status, "updated_at": now.isoformat()} |
| db.add(ReturnAudit(document_id=before["id"], action="auto_apply" if status == "active" else "reference", |
| snapshot=json.dumps({"before": before, "after": after}, ensure_ascii=False))) |
| for status, ids in groups.items(): |
| if ids: |
| db.execute(update(ReturnDocument).where(ReturnDocument.id.in_(ids), ReturnDocument.status == "pending").values(status=status, updated_at=now)) |
| if documents: |
| db.commit() |
| return {status: len(ids) for status, ids in groups.items()} |
|
|
