1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
import mysql.connector
import re
import ast
import json
from pathlib import Path
# ===== DB から実スキーマを取得 =====
def load_db_schema():
conn = mysql.connector.connect(
host="localhost",
user="mangas",
password="mangas0000",
database="mangas"
)
cur = conn.cursor()
schema = {}
cur.execute("""
SELECT table_name, column_name
FROM information_schema.columns
WHERE table_schema = DATABASE()
""")
for table, col in cur.fetchall():
schema.setdefault(table, set()).add(col)
cur.close()
conn.close()
return schema
# ===== db_service.py のコードを解析 =====
def extract_sql_strings(py_code):
"""
db_service.py 内の SQL 文を抽出する
"""
sql_strings = []
tree = ast.parse(py_code)
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
text = node.value.strip()
if ("SELECT" in text.upper() or
"INSERT" in text.upper() or
"UPDATE" in text.upper() or
"DELETE" in text.upper()):
sql_strings.append((node.lineno, text))
return sql_strings
def extract_table_and_columns(sql):
"""
SQL 文からテーブル名とカラム名を抽出する（簡易版）
"""
tables = set()
columns = set()
# FROM / JOIN
for m in re.finditer(r"\bFROM\s+([a-zA-Z0-9_]+)", sql, re.IGNORECASE):
tables.add(m.group(1))
for m in re.finditer(r"\bJOIN\s+([a-zA-Z0-9_]+)", sql, re.IGNORECASE):
tables.add(m.group(1))
# INSERT INTO table (col1, col2, ...)
m = re.search(r"INSERT\s+INTO\s+([a-zA-Z0-9_]+)\s*\((.*?)\)", sql, re.IGNORECASE | re.DOTALL)
if m:
tables.add(m.group(1))
cols = m.group(2).split(",")
for c in cols:
columns.add(c.strip())
# UPDATE table SET col = ...
m = re.search(r"UPDATE\s+([a-zA-Z0-9_]+)\s+SET\s+(.*?)\bWHERE\b", sql, re.IGNORECASE | re.DOTALL)
if m:
tables.add(m.group(1))
set_part = m.group(2)
for c in re.findall(r"([a-zA-Z0-9_]+)\s*=", set_part):
columns.add(c.strip())
# SELECT col1, col2 FROM
m = re.search(r"SELECT\s+(.*?)\bFROM\b", sql, re.IGNORECASE | re.DOTALL)
if m:
select_part = m.group(1)
for c in select_part.split(","):
c = c.strip()
if "." in c:
c = c.split(".")[1]
if c not in ("*", ""):
columns.add(c)
return tables, columns
# ===== 矛盾チェック =====
def check_consistency(db_schema, sql_entries):
errors = []
for lineno, sql in sql_entries:
tables, cols = extract_table_and_columns(sql)
for t in tables:
if t not in db_schema:
errors.append((lineno, t, None, "テーブルが存在しない"))
continue
for c in cols:
if c not in db_schema[t]:
errors.append((lineno, t, c, "カラムが存在しない"))
return errors
# ===== メイン処理 =====
def main():
db_schema = load_db_schema()
# db_schema_checker.py の場所から見た db_service.py の絶対パス
code_path = Path(__file__).resolve().parent.parent / "services" / "db_service.py"
code = code_path.read_text(encoding="utf-8")
sql_entries = extract_sql_strings(code)
errors = check_consistency(db_schema, sql_entries)
if not errors:
print("✔ 矛盾なし（db_service.py と DB スキーマは一致）")
return
print("❌ 矛盾検出:")
for lineno, table, col, msg in errors:
print(f" 行 {lineno}: {msg} → table={table}, column={col}")
if __name__ == "__main__":
main()
|
