Spaces:
Runtime error
Runtime error
File size: 4,039 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """Persisted in-memory report jobs for legacy UI compatibility."""
from __future__ import annotations
import json
import threading
import time
import uuid
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
from backend.utils import tenant_store
_lock = threading.RLock()
@dataclass
class UploadedDocument:
document_id: str
filename: str
status: str = "complete"
error: str | None = None
ingested_chunks: int = 0
storage_path: str = ""
file_size: int = 0
created_at: float = field(default_factory=time.time)
@dataclass
class ReportSession:
report_id: str
tenant_id: str
draft_id: str
survey_level: int = 3
primary_document_id: str = ""
document_ids: list[str] = field(default_factory=list)
status: str = "idle"
error_message: str | None = None
property_type: str = "semi-detached house"
tenure: str = "freehold"
generation_section_total: int | None = None
generation_started_at: float | None = None
sections_payload: dict[str, dict[str, Any]] = field(default_factory=dict)
interference_level: str = "medium"
created_at: float = field(default_factory=time.time)
updated_at: float = field(default_factory=time.time)
def _documents_path(tenant_id: str) -> Path:
return tenant_store.tenant_root(tenant_id) / "compat_documents.json"
def _reports_dir(tenant_id: str) -> Path:
d = tenant_store.tenant_root(tenant_id) / "compat_reports"
d.mkdir(parents=True, exist_ok=True)
return d
def _report_path(tenant_id: str, report_id: str) -> Path:
return _reports_dir(tenant_id) / f"{report_id}.json"
def save_document(tenant_id: str, doc: UploadedDocument) -> None:
with _lock:
path = _documents_path(tenant_id)
rows: dict[str, dict] = {}
if path.is_file():
rows = json.loads(path.read_text(encoding="utf-8"))
rows[doc.document_id] = asdict(doc)
path.write_text(json.dumps(rows, indent=2), encoding="utf-8")
def list_documents(tenant_id: str) -> dict[str, UploadedDocument]:
path = _documents_path(tenant_id)
if not path.is_file():
return {}
raw = json.loads(path.read_text(encoding="utf-8"))
return {k: UploadedDocument(**v) for k, v in raw.items()}
def get_document(tenant_id: str, document_id: str) -> UploadedDocument | None:
return list_documents(tenant_id).get(document_id)
def delete_document(tenant_id: str, document_id: str) -> bool:
with _lock:
path = _documents_path(tenant_id)
if not path.is_file():
return False
rows: dict[str, dict] = json.loads(path.read_text(encoding="utf-8"))
if document_id not in rows:
return False
del rows[document_id]
path.write_text(json.dumps(rows, indent=2), encoding="utf-8")
return True
def save_session(session: ReportSession) -> None:
session.updated_at = time.time()
with _lock:
path = _report_path(session.tenant_id, session.report_id)
path.write_text(json.dumps(asdict(session), indent=2), encoding="utf-8")
def load_session(tenant_id: str, report_id: str) -> ReportSession | None:
path = _report_path(tenant_id, report_id)
if not path.is_file():
return None
with _lock:
data = json.loads(path.read_text(encoding="utf-8"))
return ReportSession(**data)
def create_report_session(
tenant_id: str,
*,
survey_level: int,
primary_document_id: str,
document_ids: list[str],
) -> ReportSession:
from backend.core.rics_canonical_l3 import PARENT_SECTION_COUNT
session = ReportSession(
report_id=uuid.uuid4().hex,
tenant_id=tenant_id,
draft_id=uuid.uuid4().hex,
survey_level=survey_level,
primary_document_id=primary_document_id,
document_ids=document_ids,
generation_section_total=PARENT_SECTION_COUNT,
)
save_session(session)
return session
def new_document_id() -> str:
return uuid.uuid4().hex
|