fix(1c): age generated business event exports for same-cycle ingest

ansible/deploy_file_1c_analytics.yml

@@ -164,6 +164,8 @@
           landing:
             documents: {{ aw_file_1c_root }}/landing/documents
             postings: {{ aw_file_1c_root }}/landing/postings
+            business_events: {{ aw_file_1c_root }}/landing/business_events
+            document_changes: {{ aw_file_1c_root }}/landing/document_changes
             companies: {{ aw_file_1c_root }}/landing/companies
             reglog: {{ aw_file_1c_root }}/landing/reglog
             audit: {{ aw_file_1c_root }}/landing/audit
@@ -173,6 +175,8 @@
             default: jsonl
             documents: jsonl
             postings: jsonl
+            business_events: jsonl
+            document_changes: jsonl
             companies: jsonl
             reglog: jsonl
             audit: jsonl

clickhouse-1c/etl/build_business_event_exports.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 import argparse
 import hashlib
 import json
+import os
 import re
 from dataclasses import dataclass
 from datetime import UTC, datetime
@@ -90,7 +91,7 @@ def stable_id(prefix: str, *parts: Any) -> str:
     return f"{prefix}:{digest}"
 
 
-def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
+def write_jsonl(path: Path, rows: list[dict[str, Any]], *, min_age_seconds: int) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     if not rows:
         path.unlink(missing_ok=True)
@@ -99,6 +100,8 @@ def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
     payload = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n"
     tmp.write_text(payload, encoding="utf-8")
     tmp.replace(path)
+    aged_mtime = max(0, int(datetime.now(UTC).timestamp()) - max(min_age_seconds + 1, 5))
+    os.utime(path, (aged_mtime, aged_mtime))
 
 
 def load_company_index(conf: Config) -> dict[str, str]:
@@ -323,19 +326,19 @@ def main() -> int:
     for path in iter_dataset_files(conf, "documents"):
         rows = iter_rows(path, source_format(conf, "documents"))
         out = output_path(conf, "business_events", path, "business-events-documents")
-        write_jsonl(out, build_document_events(rows, path.name, company_index))
+        write_jsonl(out, build_document_events(rows, path.name, company_index), min_age_seconds=conf.min_file_age_seconds)
         print(f"built business_events from documents: {path.name} rows={len(rows)}")
 
     for path in iter_dataset_files(conf, "postings"):
         rows = iter_rows(path, source_format(conf, "postings"))
         out = output_path(conf, "business_events", path, "business-events-postings")
-        write_jsonl(out, build_posting_events(rows, path.name, company_index, by_id, by_number))
+        write_jsonl(out, build_posting_events(rows, path.name, company_index, by_id, by_number), min_age_seconds=conf.min_file_age_seconds)
         print(f"built business_events from postings: {path.name} rows={len(rows)}")
 
     for path in iter_dataset_files(conf, "audit"):
         rows = iter_rows(path, source_format(conf, "audit"))
         out = output_path(conf, "document_changes", path, "document-changes-audit")
-        write_jsonl(out, build_document_changes(rows, path.name, company_index, by_id, by_number))
+        write_jsonl(out, build_document_changes(rows, path.name, company_index, by_id, by_number), min_age_seconds=conf.min_file_age_seconds)
         print(f"built document_changes from audit: {path.name} rows={len(rows)}")
 
     return 0

clickhouse-1c/etl/test_build_business_event_exports.py

@@ -2,6 +2,8 @@
 from __future__ import annotations
 
 import sys
+import tempfile
+import time
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
@@ -99,6 +101,14 @@ class BuildBusinessEventExportsTests(unittest.TestCase):
         self.assertEqual(changes[0]["change_kind"], "repost")
         self.assertEqual(changes[0]["risk_tag"], "repost")
 
+    def test_write_jsonl_ages_generated_file(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "events.jsonl"
+            before = time.time()
+            builder.write_jsonl(path, [{"event_id": "1"}], min_age_seconds=180)
+            self.assertTrue(path.exists())
+            self.assertLessEqual(path.stat().st_mtime, before - 5)
+
 
 if __name__ == "__main__":
     unittest.main()