(user_id: str)
| 712 | candidate_tables = [table for table in ("profiles", "behavior_logs", "wiki_nodes") if table_exists(table)] |
| 713 | if not candidate_tables: |
| 714 | return [] |
| 715 | query = "\nUNION\n".join(f"SELECT user_id FROM {table}" for table in candidate_tables) |
| 716 | conn = db_ops.get_connection() |
| 717 | rows = conn.execute(f"{query}\nORDER BY user_id").fetchall() |
| 718 | conn.close() |
| 719 | return [str(row["user_id"]) for row in rows if str(row["user_id"] or "").strip()] |
| 720 | |
| 721 | |
| 722 | def stats(user_id: str) -> Dict[str, Any]: |
| 723 | init_wiki_schema() |
| 724 | conn = db_ops.get_connection() |
| 725 | cursor = conn.cursor() |
| 726 | node_rows = cursor.execute( |
| 727 | """ |
| 728 | SELECT node_type, COUNT(*) AS count |
| 729 | FROM wiki_nodes |
| 730 | WHERE user_id = ? |
| 731 | GROUP BY node_type |
| 732 | """, |
| 733 | (user_id,), |
| 734 | ).fetchall() |
| 735 | edge_count = cursor.execute( |
| 736 | "SELECT COUNT(*) AS count FROM wiki_edges WHERE user_id = ?", |
| 737 | (user_id,), |
| 738 | ).fetchone()["count"] |
| 739 | citation_count = cursor.execute( |
| 740 | "SELECT COUNT(*) AS count FROM wiki_citations WHERE user_id = ?", |
| 741 | (user_id,), |
| 742 | ).fetchone()["count"] |
| 743 | latest = cursor.execute( |
| 744 | "SELECT MAX(updated_at) AS latest FROM wiki_nodes WHERE user_id = ?", |
| 745 | (user_id,), |
| 746 | ).fetchone()["latest"] |
| 747 | conn.close() |
| 748 | by_type = {row["node_type"]: row["count"] for row in node_rows} |
| 749 | return { |
| 750 | "user_id": user_id, |
| 751 | "nodes": sum(by_type.values()), |
| 752 | "nodes_by_type": by_type, |
nothing calls this directly
no test coverage detected