261 lines
12 KiB
Python
261 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import argparse
|
|
import csv
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from typing import List, Optional
|
|
|
|
|
|
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if PROJECT_ROOT not in sys.path:
|
|
sys.path.insert(0, PROJECT_ROOT)
|
|
|
|
from analytics import AnalyticsRepository, _classify_restriction_status, _make_task_key
|
|
from monitoring import MonitoringRepository
|
|
|
|
|
|
def _table_columns(connection: sqlite3.Connection, table_name: str) -> List[str]:
|
|
return [row[1] for row in connection.execute(f"PRAGMA table_info({table_name})").fetchall()]
|
|
|
|
|
|
def import_monitoring(old_db_path: str, target_db_path: str):
|
|
MonitoringRepository(db_path=target_db_path)
|
|
AnalyticsRepository(db_path=target_db_path)
|
|
|
|
with sqlite3.connect(old_db_path) as source, sqlite3.connect(target_db_path) as target:
|
|
source.row_factory = sqlite3.Row
|
|
target.row_factory = sqlite3.Row
|
|
source_columns = set(_table_columns(source, "task_execution"))
|
|
target_columns = set(_table_columns(target, "task_execution"))
|
|
common_columns = sorted((source_columns & target_columns) - {"execution_id"})
|
|
select_columns = ["execution_id", *common_columns]
|
|
placeholders = ", ".join(["?"] * len(select_columns))
|
|
insert_columns = ", ".join(select_columns)
|
|
update_columns = ", ".join([f"{column}=excluded.{column}" for column in common_columns])
|
|
|
|
rows = source.execute(f"SELECT {insert_columns} FROM task_execution").fetchall()
|
|
for row in rows:
|
|
target.execute(
|
|
f"""
|
|
INSERT INTO task_execution ({insert_columns})
|
|
VALUES ({placeholders})
|
|
ON CONFLICT(execution_id) DO UPDATE SET {update_columns}
|
|
""",
|
|
[row[column] for column in select_columns],
|
|
)
|
|
target.commit()
|
|
print(f"Imported {len(rows)} task_execution rows into {target_db_path}")
|
|
|
|
|
|
def _parse_bool(value: str) -> bool:
|
|
return str(value or "").strip().lower() in {"1", "true", "yes"}
|
|
|
|
|
|
def _split_names(raw_value: str) -> List[str]:
|
|
return [item.strip() for item in str(raw_value or "").split(",") if item.strip()]
|
|
|
|
|
|
def _resolve_self_ratio(self_ratio: object, self_traffic_bytes: object = 0, total_traffic_bytes: object = 0) -> float:
|
|
try:
|
|
normalized_ratio = float(self_ratio or 0.0)
|
|
except (TypeError, ValueError):
|
|
normalized_ratio = 0.0
|
|
if normalized_ratio > 0:
|
|
return normalized_ratio
|
|
try:
|
|
normalized_self_bytes = float(self_traffic_bytes or 0.0)
|
|
except (TypeError, ValueError):
|
|
normalized_self_bytes = 0.0
|
|
try:
|
|
normalized_total_bytes = float(total_traffic_bytes or 0.0)
|
|
except (TypeError, ValueError):
|
|
normalized_total_bytes = 0.0
|
|
if normalized_self_bytes > 0 and normalized_total_bytes > 0:
|
|
return round((normalized_self_bytes / normalized_total_bytes) * 100, 2)
|
|
return 0.0
|
|
|
|
|
|
def import_app_summary(summary_csv_path: str, target_db_path: str):
|
|
repo = AnalyticsRepository(db_path=target_db_path)
|
|
imported = 0
|
|
with open(summary_csv_path, "r", encoding="utf-8-sig", newline="") as handle:
|
|
reader = csv.DictReader(handle)
|
|
for row in reader:
|
|
package_name = str(row.get("package_name") or "").strip()
|
|
if not package_name:
|
|
continue
|
|
unique_domains = _split_names(row.get("unique_domain_names", ""))
|
|
second_level_domains = _split_names(row.get("unique_second_level_domains", ""))
|
|
summary = {
|
|
"app_name": str(row.get("app_name") or "").strip(),
|
|
"latest_task_key": "",
|
|
"latest_status": "success" if _parse_bool(row.get("test_success", "")) else "failed",
|
|
"latest_test_time": 0.0,
|
|
"latest_worker_id": str(row.get("test_host") or "").strip(),
|
|
"latest_task_detail": str(row.get("task_detail") or "").strip(),
|
|
"latest_failure_type": str(row.get("failure_type") or "").strip(),
|
|
"unique_domain_count": len(unique_domains),
|
|
"unique_domain_names": unique_domains,
|
|
"unique_second_level_domain_count": len(second_level_domains),
|
|
"unique_second_level_domains": second_level_domains,
|
|
"droidbot_steps": int(float(row.get("droidbot_steps") or 0)),
|
|
"gui_agent_steps": int(float(row.get("gui_agent_steps") or 0)),
|
|
"duration_seconds": float(row.get("duration_seconds") or 0.0),
|
|
"num_nodes": int(float(row.get("num_nodes") or 0)),
|
|
"num_reached_activities": int(float(row.get("num_reached_activities") or 0)),
|
|
"app_num_total_activities": int(float(row.get("app_num_total_activities") or 0)),
|
|
"total_traffic_bytes": 0,
|
|
"self_traffic_bytes": 0,
|
|
"server_traffic_bytes": 0,
|
|
"unrecognized_traffic_bytes": 0,
|
|
"self_ratio": float(row.get("self_ratio") or 0.0),
|
|
"recognition_ratio": 0.0,
|
|
"artifact_status": "partial",
|
|
}
|
|
restriction_status, retryability = _classify_restriction_status(
|
|
summary["latest_status"],
|
|
_resolve_self_ratio(summary["self_ratio"], summary["self_traffic_bytes"], summary["total_traffic_bytes"]),
|
|
summary["latest_failure_type"],
|
|
)
|
|
summary["restriction_status"] = restriction_status
|
|
summary["retryability"] = retryability
|
|
repo.replace_package_snapshot(package_name, summary, [], [], [])
|
|
imported += 1
|
|
print(f"Imported {imported} app summary rows into {target_db_path}")
|
|
|
|
|
|
def _get_latest_task_fallback(connection: sqlite3.Connection, package_name: str) -> Optional[sqlite3.Row]:
|
|
task_columns = set(_table_columns(connection, "task_execution"))
|
|
if not task_columns:
|
|
return None
|
|
selected_columns = [
|
|
"status" if "status" in task_columns else "'' AS status",
|
|
"error_type" if "error_type" in task_columns else "'' AS error_type",
|
|
"result_detail" if "result_detail" in task_columns else "'' AS result_detail",
|
|
"error_message" if "error_message" in task_columns else "'' AS error_message",
|
|
"worker_id" if "worker_id" in task_columns else "'' AS worker_id",
|
|
"task_key" if "task_key" in task_columns else "'' AS task_key",
|
|
"COALESCE(task_ended_at, task_started_at, last_updated_at, 0) AS latest_time",
|
|
]
|
|
return connection.execute(
|
|
f"""
|
|
SELECT {', '.join(selected_columns)}
|
|
FROM task_execution
|
|
WHERE package_name = ?
|
|
ORDER BY COALESCE(task_ended_at, task_started_at, last_updated_at, 0) DESC, execution_id DESC
|
|
LIMIT 1
|
|
""",
|
|
(package_name,),
|
|
).fetchone()
|
|
|
|
|
|
def repair_app_summary(target_db_path: str, only_missing: bool = False):
|
|
repo = AnalyticsRepository(db_path=target_db_path)
|
|
repaired = 0
|
|
skipped = 0
|
|
with repo._connect() as connection:
|
|
rows = repo._list_catalog_summaries(connection)
|
|
for row in rows:
|
|
package_name = str(row["package_name"] or "").strip()
|
|
if not package_name:
|
|
skipped += 1
|
|
continue
|
|
if only_missing and row["restriction_status"] and row["retryability"] and row["latest_status"]:
|
|
skipped += 1
|
|
continue
|
|
|
|
latest_status = str(row["latest_status"] or "").strip()
|
|
latest_failure_type = str(row["latest_failure_type"] or "").strip()
|
|
fallback = None
|
|
if not latest_status or not latest_failure_type:
|
|
fallback_row = _get_latest_task_fallback(connection, package_name)
|
|
fallback = dict(fallback_row) if fallback_row else None
|
|
if fallback:
|
|
latest_status = latest_status or str(fallback["status"] or "").strip()
|
|
latest_failure_type = latest_failure_type or str(fallback["error_type"] or "").strip()
|
|
if not latest_status and not latest_failure_type:
|
|
skipped += 1
|
|
continue
|
|
|
|
app_name = str(row.get("app_name") or package_name).strip() or package_name
|
|
repo.replace_package_snapshot(
|
|
package_name,
|
|
{
|
|
"app_name": app_name,
|
|
"app_magic_label": row.get("app_magic_label", ""),
|
|
"latest_task_key": str((fallback or {}).get("task_key") or row.get("latest_task_key") or _make_task_key(app_name, package_name)).strip(),
|
|
"latest_status": latest_status,
|
|
"latest_worker_id": str((fallback or {}).get("worker_id") or row.get("latest_worker_id") or "").strip(),
|
|
"latest_failure_type": latest_failure_type,
|
|
"latest_task_detail": str(
|
|
(fallback or {}).get("result_detail")
|
|
or (fallback or {}).get("error_message")
|
|
or row.get("latest_task_detail")
|
|
or ""
|
|
).strip(),
|
|
"latest_test_time": float((fallback or {}).get("latest_time") or row.get("latest_test_time") or 0.0),
|
|
"collection_status_reason": latest_failure_type or latest_status,
|
|
"collection_task_type": row.get("collection_task_type") or "new_app",
|
|
"duration_seconds": row.get("duration_seconds", 0),
|
|
"droidbot_steps": row.get("droidbot_steps", 0),
|
|
"gui_agent_steps": row.get("gui_agent_steps", 0),
|
|
"num_nodes": row.get("num_nodes", 0),
|
|
"num_reached_activities": row.get("num_reached_activities", 0),
|
|
"app_num_total_activities": row.get("app_num_total_activities", 0),
|
|
"total_traffic_bytes": row.get("total_traffic_bytes", 0),
|
|
"self_traffic_bytes": row.get("self_traffic_bytes", 0),
|
|
"server_traffic_bytes": row.get("server_traffic_bytes", 0),
|
|
"unrecognized_traffic_bytes": row.get("unrecognized_traffic_bytes", 0),
|
|
"model_flow_count": row.get("model_flow_count", 0),
|
|
"model_traffic_bytes": row.get("model_traffic_bytes", 0),
|
|
"task_payload": row.get("task_payload") or {},
|
|
"downloads": row.get("downloads"),
|
|
"source_order": row.get("source_order"),
|
|
},
|
|
[],
|
|
[],
|
|
[],
|
|
)
|
|
repaired += 1
|
|
suffix = " (only missing rows)" if only_missing else ""
|
|
print(f"Repaired {repaired} app_catalog/collection_task rows{suffix} in {target_db_path}; skipped {skipped}")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Seed analytics tables from legacy files")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
monitoring_parser = subparsers.add_parser("import-monitoring", help="Import task_execution rows from a legacy monitoring sqlite db")
|
|
monitoring_parser.add_argument("--from", dest="from_path", required=True, help="Legacy monitoring sqlite path")
|
|
monitoring_parser.add_argument("--to", dest="to_path", required=True, help="Target runtime sqlite path")
|
|
|
|
summary_parser = subparsers.add_parser("import-app-summary", help="Import legacy app_domain_summary.csv into app_catalog and collection_task")
|
|
summary_parser.add_argument("--from", dest="from_path", required=True, help="Legacy app_domain_summary.csv path")
|
|
summary_parser.add_argument("--to", dest="to_path", required=True, help="Target runtime sqlite path")
|
|
|
|
repair_parser = subparsers.add_parser("repair-app-summary", help="Backfill missing latest collection_task rows from task_execution")
|
|
repair_parser.add_argument("--to", dest="to_path", required=True, help="Target runtime sqlite path")
|
|
repair_parser.add_argument(
|
|
"--only-missing",
|
|
action="store_true",
|
|
help="Only repair rows where restriction_status/retryability are still empty",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main():
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
if args.command == "import-monitoring":
|
|
import_monitoring(args.from_path, args.to_path)
|
|
elif args.command == "import-app-summary":
|
|
import_app_summary(args.from_path, args.to_path)
|
|
elif args.command == "repair-app-summary":
|
|
repair_app_summary(args.to_path, only_missing=bool(args.only_missing))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|