modules/registro-imprese-sari/scripts/initialize_case.py

"""Initialize private draft files for a Registro Imprese/SARI case."""

from __future__ import annotations

import argparse
import logging
from pathlib import Path

from case_core import (
    PLUGIN_NAME,
    AssuranceContractError,
    ensure_safe_output_dir,
    iso_now,
    load_running_case_context,
    safe_identifier,
    validate_iso_date,
    write_private_json,
)
from registry_display import text

__all__ = ["initialize_case", "main"]

LOGGER = logging.getLogger(__name__)
PLUGIN_ROOT = Path(__file__).resolve().parents[1]


def initialize_case(
    output_dir: Path,
    *,
    run_id: str,
    reference_date: str,
    client_reference: str,
    language: str = "it",
    jurisdiction: str = "IT",
    client_engagement: Path | None = None,
) -> dict[str, Path]:
    """Create bounded empty drafts without choosing legal classifications."""

    if jurisdiction not in {"IT", "CH-GE"}:
        raise ValueError("Supported registry jurisdictions are IT and CH-GE")
    run_id = safe_identifier(run_id, field="run_id")
    client_reference = safe_identifier(client_reference, field="client_reference")
    reference_date = validate_iso_date(reference_date, field="reference_date")
    language = str(language or "").strip().lower()
    if not 2 <= len(language) <= 12 or not language.replace("-", "").isalpha():
        raise ValueError("language must be a short language tag")
    safe_output = ensure_safe_output_dir(output_dir, plugin_root=PLUGIN_ROOT)
    output_reference = safe_output.as_posix()
    context_reference = (
        client_engagement.resolve().as_posix()
        if client_engagement is not None
        else None
    )
    path_reference = "absolute"
    if client_engagement is not None:
        context = load_running_case_context(
            client_engagement,
            output_dir=safe_output,
        )
        if str(context["run_id"]) != run_id:
            raise ValueError("run_id must match the customer-folder workflow run")
        output_reference = safe_output.relative_to(Path(context["run_root"])).as_posix()
        context_reference = Path(client_engagement).name
        path_reference = "run_root_relative"
    intake_path = safe_output / "case_intake_draft.json"
    plan_path = safe_output / "practice_plan_draft.json"
    run_path = safe_output / "run_intake.json"
    if any(path.exists() for path in (intake_path, plan_path, run_path)):
        raise FileExistsError("case draft files already exist; do not overwrite them")
    write_private_json(
        intake_path,
        {
            "schema_version": "1.0",
            "plugin": PLUGIN_NAME,
            "jurisdiction": jurisdiction,
            "run_id": run_id,
            "reference_date": reference_date,
            "client_reference": client_reference,
            "client_identity": {
                "name": "",
                "tax_code": "",
                "vat_number": "",
                "email": "",
                "pec": "",
                "phone": "",
                "address": "",
            },
            "competent_chamber": {
                "tenant": "",
                "name": "",
                "territorial_basis": "",
                "confirmation_status": "unknown",
            },
            "subject": {
                "legal_form": "",
                "confirmation_status": "unknown",
            },
            "activity": {
                "description": "",
                "classification_status": "unknown",
                "ateco_proposal": None,
            },
            "requested_operation": {
                "description": "",
                "position_types": [],
                "effective_date": None,
                "confirmation_status": "unknown",
            },
            "current_positions": [],
            "professional_question": "",
        },
    )
    write_private_json(
        plan_path,
        {
            "schema_version": "1.0",
            "plugin": PLUGIN_NAME,
            "jurisdiction": jurisdiction,
            "run_id": run_id,
            "case_summary": "",
            "review_context": {},
            "classification_proposals": [],
            "position_matrix": [],
            "dire_steps": [],
            "required_documents": [],
            "application_fields": [],
            "risks": [],
            "missing_information": [],
            "sari_question_draft": "",
            "limitations": [],
            "professional_review": {
                "status": "pending",
                "reviewer_id": None,
                "reviewer_role": None,
                "reviewed_at": None,
                "notes": None,
            },
        },
    )
    write_private_json(
        run_path,
        {
            "schema_version": "1.0",
            "plugin": PLUGIN_NAME,
            "jurisdiction": jurisdiction,
            "workflow": PLUGIN_NAME,
            "run_id": run_id,
            "reference_date": reference_date,
            "created_at": iso_now(),
            "language": language,
            "path_reference": path_reference,
            "input_paths": [],
            "output_dir": output_reference,
            "inferred_task": (text(language, "initial_task")),
            "assumptions": [text(language, "initial_assumption")],
            "unresolved_questions": [text(language, "initial_question")],
            "dependency_check": {
                "status": "not_recorded",
                "command": ["python", "scripts/check_dependencies.py"],
                "requirements": ["requirements.txt"],
            },
            "status": "pending",
            "data_posture": {
                "local_files_read": [],
                "external_connectors_used": [],
                "external_routes_used": [],
                "upload_paths_used": [],
                "hosted_notebook_execution_used": False,
                "remote_sql_execution_used": False,
                "private_local_output": True,
                "private_case_details_available_for_professional_review": True,
                "public_sari_queries_require_generic_terms": True,
                "network_calls_by_default": False,
                "credentials_or_session_export": False,
                "portal_submission": False,
            },
            "execution_trace": [
                {
                    "step_id": "initialize_case",
                    "kind": "deterministic_initialization",
                    "command": [
                        "python",
                        "scripts/initialize_case.py",
                        "--output-dir",
                        output_reference,
                        *(
                            [
                                "--client-engagement",
                                context_reference,
                            ]
                            if client_engagement is not None
                            else []
                        ),
                        "--reference-date",
                        reference_date,
                        "--client-reference",
                        client_reference,
                        "--language",
                        language,
                    ],
                    "execution_location": "local_python",
                    "status": "passed",
                    "inputs": [],
                    "outputs": [intake_path.name, plan_path.name, run_path.name],
                }
            ],
        },
    )
    return {"intake": intake_path, "plan": plan_path, "run_intake": run_path}


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument("--client-engagement", type=Path, required=True)
    parser.add_argument("--reference-date", required=True)
    parser.add_argument("--client-reference", required=True)
    parser.add_argument("--language", default="it")
    parser.add_argument("--jurisdiction", choices=("IT", "CH-GE"), default="IT")
    args = parser.parse_args(argv)
    try:
        context = load_running_case_context(
            args.client_engagement,
            output_dir=args.output_dir,
        )
        paths = initialize_case(
            args.output_dir,
            run_id=context["run_id"],
            reference_date=args.reference_date,
            client_reference=args.client_reference,
            language=args.language,
            jurisdiction=args.jurisdiction,
            client_engagement=args.client_engagement,
        )
    except (AssuranceContractError, OSError, ValueError) as exc:
        LOGGER.error("INITIALIZATION_BLOCKED: %s", exc)
        return 2
    LOGGER.info("Initialized case drafts in %s", paths["run_intake"].parent)
    return 0


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    raise SystemExit(main())