rmancinas
0268ed896c
ODBC only exposes Tables and non-hidden SELECT queries, so this used DAO COM automation (migration/catalog_objects.py, requires pywin32) to catalog Reports, Forms, and full Query SQL text across all four files instead. Key finding: SEGUROS 16.mdb, previously noted as having zero data tables, turns out to hold all 212 Reports/149 Forms/857 Queries for the insurance line - SEGUROS 16_be.mdb is confirmed pure data storage with zero saved objects. The renewal-notice reports also reveal a RENEW/RENEW2/RENEW3 multi-notice reminder cycle not visible in the table schema. docs/LEGACY_DATABASES_OBJECTS.md documents all of this with the full 751 real queries' SQL text (business logic: billing math, year- rollover batches, duplicate/delinquency detection). Raw output at migration/objects.json. Cross-linked from RESUME.md and the existing table-only LEGACY_DATABASES.md.
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
"""
|
|
Catalogs the non-table Access objects (Queries, Forms, Reports) via DAO COM
|
|
automation (win32com), since ODBC only exposes Tables and a subset of
|
|
non-hidden SELECT queries (as "VIEW"). DAO exposes everything, including:
|
|
- QueryDefs: every saved query, with full .SQL text (this is where a lot
|
|
of the business logic actually lives - billing calculations, renewal
|
|
schedules, report data sources)
|
|
- Containers("Reports").Documents: every saved Report object (name only -
|
|
DAO's catalog interface doesn't expose RecordSource/layout; that needs
|
|
the full Access.Application object model, which isn't available without
|
|
MS Access itself installed)
|
|
- Containers("Forms").Documents: every saved Form object (name only, same
|
|
limitation)
|
|
|
|
Requires: pywin32 (`pip install pywin32`) and the Access Database Engine
|
|
(same requirement as everything else in migration/ - Windows only).
|
|
|
|
Usage:
|
|
python catalog_objects.py --output objects.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import win32com.client
|
|
|
|
from config import SOURCE_ROOT
|
|
|
|
HIDDEN_QUERY_PREFIXES = ("~sq_", "~TMP")
|
|
|
|
# Unlike config.SOURCES (which points "seguros" at the _be data file), the
|
|
# Queries/Forms/Reports objects live in the linked *frontend* files - the
|
|
# _be backend is pure data storage. Both SEGUROS files are catalogued here
|
|
# since either could theoretically hold saved objects.
|
|
OBJECT_SOURCES = {
|
|
"utilities": SOURCE_ROOT / "UTILITIES.accdb",
|
|
"seguros_frontend": SOURCE_ROOT / "SEGUROS 16.mdb",
|
|
"seguros_backend": SOURCE_ROOT / "SEGUROS 16_be.mdb",
|
|
"scothia": SOURCE_ROOT / "SCOTHIA.mdb",
|
|
}
|
|
|
|
|
|
def catalog_source(path: Path) -> dict:
|
|
engine = win32com.client.Dispatch("DAO.DBEngine.120")
|
|
db = engine.OpenDatabase(str(path))
|
|
|
|
queries = []
|
|
for i in range(db.QueryDefs.Count):
|
|
qd = db.QueryDefs(i)
|
|
name = qd.Name
|
|
is_hidden = name.startswith(HIDDEN_QUERY_PREFIXES)
|
|
try:
|
|
sql = qd.SQL
|
|
except Exception as exc: # noqa: BLE001
|
|
sql = f"<could not read SQL: {exc}>"
|
|
try:
|
|
query_type = qd.Type # 0=Select, 32=CrossTab, 48=Delete, 64=Update, 80=Append, 96=MakeTable, 112=DDL, 128=PassThrough, 144=SPTBulk, 160=SetOperation, 176=Procedure
|
|
except Exception:
|
|
query_type = None
|
|
queries.append({"name": name, "hidden": is_hidden, "sql": sql, "type": query_type})
|
|
|
|
def list_container(container_name: str) -> list[str]:
|
|
container = db.Containers(container_name)
|
|
names = []
|
|
for i in range(container.Documents.Count):
|
|
names.append(container.Documents(i).Name)
|
|
return sorted(names)
|
|
|
|
reports = list_container("Reports")
|
|
forms = list_container("Forms")
|
|
|
|
db.Close()
|
|
|
|
return {
|
|
"query_count": len(queries),
|
|
"queries": queries,
|
|
"report_count": len(reports),
|
|
"reports": reports,
|
|
"form_count": len(forms),
|
|
"forms": forms,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--output", type=Path, default=Path("objects.json"))
|
|
args = parser.parse_args()
|
|
|
|
catalog = {}
|
|
for source_name, path in OBJECT_SOURCES.items():
|
|
print(f"Cataloging objects in {source_name} ({path})...")
|
|
if not Path(path).exists():
|
|
print(f" [skip] not found")
|
|
continue
|
|
catalog[source_name] = catalog_source(path)
|
|
c = catalog[source_name]
|
|
print(f" queries={c['query_count']} reports={c['report_count']} forms={c['form_count']}")
|
|
|
|
args.output.write_text(json.dumps(catalog, indent=2, default=str), encoding="utf-8")
|
|
print(f"Wrote {args.output}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|