135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Query the CMA SQLite database at /root/cma-management/backend/cma.db
|
|
Extracts all records from actual_costs, standard_costs, and checks kpi_values/budget_plans.
|
|
Run: python3 scripts/query_cma_db.py
|
|
"""
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
DB_PATH = "/root/cma-management/backend/cma.db"
|
|
|
|
def print_header(title):
|
|
print(f"\n{'='*100}")
|
|
print(f" {title}")
|
|
print(f"{'='*100}")
|
|
|
|
def print_table_info(conn):
|
|
tables = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
|
).fetchall()
|
|
print_header("CMA DATABASE — ALL TABLES")
|
|
for t in tables:
|
|
name = t[0]
|
|
cnt = conn.execute(f"SELECT COUNT(*) FROM \"{name}\"").fetchone()[0]
|
|
cols = conn.execute(f"PRAGMA table_info(\"{name}\")").fetchall()
|
|
col_str = ", ".join(f"{c[1]}({c[2]})" for c in cols)
|
|
print(f" 📦 {name:30s} ({cnt} rows) [{col_str}]")
|
|
|
|
def query_all(sql, desc, conn):
|
|
print_header(desc)
|
|
cursor = conn.execute(sql)
|
|
rows = cursor.fetchall()
|
|
print(f" SQL: {sql}")
|
|
print(f" Rows returned: {len(rows)}")
|
|
print()
|
|
if rows:
|
|
headers = [d[0] for d in cursor.description]
|
|
# Column widths
|
|
widths = {}
|
|
for i, h in enumerate(headers):
|
|
widths[i] = max(len(str(h)), 10)
|
|
for r in rows:
|
|
for i, v in enumerate(r):
|
|
widths[i] = max(widths[i], len(str(v)) if v is not None else 4)
|
|
|
|
# Header row
|
|
sep = " | "
|
|
hdr_line = sep.join(h.ljust(widths[i]) for i, h in enumerate(headers))
|
|
print(f" {hdr_line}")
|
|
print(f" {'-' * len(hdr_line)}")
|
|
|
|
for r in rows:
|
|
val_line = sep.join(
|
|
(str(r[i]) if r[i] is not None else "NULL").ljust(widths[i])
|
|
for i in range(len(headers))
|
|
)
|
|
print(f" {val_line}")
|
|
else:
|
|
print(" (no data)")
|
|
print()
|
|
|
|
def main():
|
|
try:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
except sqlite3.Error as e:
|
|
print(f"ERROR: Cannot connect to database at {DB_PATH}: {e}")
|
|
sys.exit(1)
|
|
|
|
print(f"🔍 CMA Database Query Report")
|
|
print(f" Database: {DB_PATH}")
|
|
print(f" Timestamp: {datetime.now().isoformat()}")
|
|
|
|
# (1) All tables
|
|
print_table_info(conn)
|
|
|
|
# (2) actual_costs — all records
|
|
query_all(
|
|
"SELECT * FROM actual_costs ORDER BY period, product_code, cost_type",
|
|
"REQUIRED (1) — actual_costs: ALL RECORDS",
|
|
conn
|
|
)
|
|
|
|
# (3) standard_costs — all records
|
|
query_all(
|
|
"SELECT * FROM standard_costs ORDER BY product_code, cost_type, item_name",
|
|
"REQUIRED (2) — standard_costs: ALL RECORDS",
|
|
conn
|
|
)
|
|
|
|
# (4) Check kpi_values
|
|
cnt_kpi = conn.execute("SELECT COUNT(*) FROM kpi_values").fetchone()[0]
|
|
print_header(f"REQUIRED (3a) — kpi_values: {cnt_kpi} rows")
|
|
if cnt_kpi > 0:
|
|
print(" DATA EXISTS")
|
|
query_all(
|
|
"SELECT * FROM kpi_values ORDER BY period, kpi_id LIMIT 50",
|
|
f"kpi_values data (showing up to 50 rows of {cnt_kpi})",
|
|
conn
|
|
)
|
|
else:
|
|
print(" ⚠️ NO DATA — table exists but is empty")
|
|
|
|
# (5) Check budget_plans
|
|
cnt_budget = conn.execute("SELECT COUNT(*) FROM budget_plans").fetchone()[0]
|
|
print_header(f"REQUIRED (3b) — budget_plans: {cnt_budget} rows")
|
|
if cnt_budget > 0:
|
|
print(" DATA EXISTS")
|
|
query_all(
|
|
"SELECT * FROM budget_plans ORDER BY period, product_code LIMIT 50",
|
|
f"budget_plans data (showing up to 50 rows of {cnt_budget})",
|
|
conn
|
|
)
|
|
else:
|
|
print(" ⚠️ NO DATA — table exists but is empty")
|
|
|
|
# Bonus: Schema details for the requested tables
|
|
for tbl in ['actual_costs', 'standard_costs', 'kpi_values', 'budget_plans']:
|
|
cols = conn.execute(f"PRAGMA table_info(\"{tbl}\")").fetchall()
|
|
cnt = conn.execute(f"SELECT COUNT(*) FROM \"{tbl}\"").fetchone()[0]
|
|
print(f"\n 📋 {tbl}: {cnt} rows")
|
|
for c in cols:
|
|
nullable = "NULL" if c[3] else "NOT NULL"
|
|
pk = "PK" if c[5] else ""
|
|
default = f"default={c[4]}" if c[4] is not None else ""
|
|
print(f" ├ {c[1]:25s} {c[2]:15s} {nullable:10s} {pk:3s} {default}")
|
|
|
|
conn.close()
|
|
print(f"\n{'='*100}")
|
|
print(" ✅ DONE")
|
|
print(f"{'='*100}\n")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|