387 lines
13 KiB
Python
387 lines
13 KiB
Python
import httplib2
|
|
import googleapiclient.discovery
|
|
from oauth2client.service_account import ServiceAccountCredentials
|
|
|
|
from config import CREDENTIALS_FILE, SPREADSHEET_ID
|
|
from database.db import get_connection
|
|
|
|
SCOPES = [
|
|
"https://www.googleapis.com/auth/spreadsheets",
|
|
"https://www.googleapis.com/auth/drive",
|
|
]
|
|
|
|
|
|
def get_google_service():
|
|
credentials = ServiceAccountCredentials.from_json_keyfile_name(
|
|
CREDENTIALS_FILE, SCOPES
|
|
)
|
|
http = credentials.authorize(httplib2.Http())
|
|
return googleapiclient.discovery.build(
|
|
"sheets", "v4", http=http, cache_discovery=False
|
|
)
|
|
|
|
|
|
def _fetch_rows(service, sheet_range: str):
|
|
return service.spreadsheets().values().get(
|
|
spreadsheetId=SPREADSHEET_ID,
|
|
range=sheet_range,
|
|
majorDimension="ROWS",
|
|
).execute().get("values", [])
|
|
|
|
|
|
def _require_columns(row, count: int, sheet_name: str, idx: int):
|
|
if len(row) < count:
|
|
raise ValueError(f"{sheet_name} row {idx}: expected {count} columns, got {len(row)}")
|
|
|
|
|
|
def _require_non_empty(value: str, sheet_name: str, idx: int, column_name: str) -> str:
|
|
value = (value or "").strip()
|
|
if not value:
|
|
raise ValueError(f"{sheet_name} row {idx}: {column_name} is empty")
|
|
return value
|
|
|
|
|
|
def _is_empty_row(row) -> bool:
|
|
return not row or not any((value or "").strip() for value in row)
|
|
|
|
|
|
def _validate_pinpad_errors(rows):
|
|
if not rows:
|
|
raise ValueError("PinPad: table is empty")
|
|
|
|
parsed = []
|
|
seen_codes = set()
|
|
for idx, row in enumerate(rows, start=2):
|
|
_require_columns(row, 3, "PinPad", idx)
|
|
code = int(_require_non_empty(row[0], "PinPad", idx, "code"))
|
|
if code in seen_codes:
|
|
raise ValueError(f"PinPad row {idx}: duplicate code {code}")
|
|
seen_codes.add(code)
|
|
parsed.append(
|
|
(
|
|
code,
|
|
_require_non_empty(row[1], "PinPad", idx, "reason"),
|
|
_require_non_empty(row[2], "PinPad", idx, "action"),
|
|
)
|
|
)
|
|
return parsed
|
|
|
|
|
|
def _validate_terminal(rows, key_rows, step_rows):
|
|
if not rows:
|
|
raise ValueError("Terminal_Instructions: table is empty")
|
|
if not key_rows:
|
|
raise ValueError("Terminal_Keys: table is empty")
|
|
if not step_rows:
|
|
raise ValueError("Terminal_Steps: table is empty")
|
|
|
|
instructions = []
|
|
instruction_ids = set()
|
|
for idx, row in enumerate(rows, start=2):
|
|
_require_columns(row, 2, "Terminal_Instructions", idx)
|
|
instruction_id = int(_require_non_empty(row[0], "Terminal_Instructions", idx, "id"))
|
|
if instruction_id in instruction_ids:
|
|
raise ValueError(f"Terminal_Instructions row {idx}: duplicate id {instruction_id}")
|
|
instruction_ids.add(instruction_id)
|
|
instructions.append(
|
|
(
|
|
instruction_id,
|
|
_require_non_empty(row[1], "Terminal_Instructions", idx, "title"),
|
|
)
|
|
)
|
|
|
|
keys = []
|
|
for idx, row in enumerate(key_rows, start=2):
|
|
_require_columns(row, 3, "Terminal_Keys", idx)
|
|
instruction_id = int(_require_non_empty(row[0], "Terminal_Keys", idx, "instruction_id"))
|
|
if instruction_id not in instruction_ids:
|
|
raise ValueError(f"Terminal_Keys row {idx}: unknown instruction_id {instruction_id}")
|
|
key_type = _require_non_empty(row[2], "Terminal_Keys", idx, "key_type").lower()
|
|
if key_type not in ("code", "text"):
|
|
raise ValueError(f"Terminal_Keys row {idx}: key_type must be code or text")
|
|
keys.append(
|
|
(
|
|
instruction_id,
|
|
_require_non_empty(row[1], "Terminal_Keys", idx, "key").lower(),
|
|
key_type,
|
|
)
|
|
)
|
|
|
|
steps = []
|
|
for idx, row in enumerate(step_rows, start=2):
|
|
if _is_empty_row(row):
|
|
continue
|
|
if len(row) < 3:
|
|
print(f"WARN Terminal_Steps row {idx}: skipped incomplete row")
|
|
continue
|
|
instruction_id = int(_require_non_empty(row[0], "Terminal_Steps", idx, "instruction_id"))
|
|
if instruction_id not in instruction_ids:
|
|
raise ValueError(f"Terminal_Steps row {idx}: unknown instruction_id {instruction_id}")
|
|
step_type = _require_non_empty(row[2], "Terminal_Steps", idx, "type").lower()
|
|
if step_type not in ("text", "image", "pause", "goto"):
|
|
raise ValueError(f"Terminal_Steps row {idx}: type must be text, image, pause or goto")
|
|
content = (row[3] if len(row) > 3 else "").strip()
|
|
if step_type != "pause" and not content:
|
|
print(f"WARN Terminal_Steps row {idx}: skipped {step_type} step with empty content")
|
|
continue
|
|
steps.append(
|
|
(
|
|
instruction_id,
|
|
int(_require_non_empty(row[1], "Terminal_Steps", idx, "step_order")),
|
|
step_type,
|
|
content,
|
|
)
|
|
)
|
|
|
|
if not steps:
|
|
raise ValueError("Terminal_Steps: no valid steps found")
|
|
|
|
return instructions, keys, steps
|
|
|
|
|
|
def _validate_tech_problems(rows):
|
|
if not rows:
|
|
raise ValueError("TechProblems: table is empty")
|
|
|
|
problems = []
|
|
problem_ids = set()
|
|
for idx, row in enumerate(rows, start=2):
|
|
_require_columns(row, 3, "TechProblems", idx)
|
|
problem_id = _require_non_empty(row[0], "TechProblems", idx, "id")
|
|
if problem_id in problem_ids:
|
|
raise ValueError(f"TechProblems row {idx}: duplicate id {problem_id}")
|
|
task_type = _require_non_empty(row[1], "TechProblems", idx, "task_type").upper()
|
|
if task_type not in ("ADMIN", "TECH"):
|
|
raise ValueError(f"TechProblems row {idx}: task_type must be ADMIN or TECH")
|
|
problem_ids.add(problem_id)
|
|
problems.append(
|
|
(
|
|
problem_id,
|
|
task_type,
|
|
_require_non_empty(row[2], "TechProblems", idx, "keywords"),
|
|
)
|
|
)
|
|
return problems, problem_ids
|
|
|
|
|
|
def _validate_tech_solutions(rows, problem_ids):
|
|
if not rows:
|
|
raise ValueError("TechProblems_Solution: table is empty")
|
|
|
|
solutions = []
|
|
seen_ids = set()
|
|
for idx, row in enumerate(rows, start=2):
|
|
_require_columns(row, 8, "TechProblems_Solution", idx)
|
|
problem_id = _require_non_empty(row[0], "TechProblems_Solution", idx, "problem_id")
|
|
if problem_id in seen_ids:
|
|
raise ValueError(f"TechProblems_Solution row {idx}: duplicate problem_id {problem_id}")
|
|
if problem_id not in problem_ids:
|
|
raise ValueError(f"TechProblems_Solution row {idx}: unknown problem_id {problem_id}")
|
|
task_type = _require_non_empty(row[2], "TechProblems_Solution", idx, "task_type").upper()
|
|
can_fix_self = _require_non_empty(row[3], "TechProblems_Solution", idx, "can_fix_self").upper()
|
|
need_result_feedback = _require_non_empty(row[4], "TechProblems_Solution", idx, "need_result_feedback").upper()
|
|
if task_type not in ("ADMIN", "TECH"):
|
|
raise ValueError(f"TechProblems_Solution row {idx}: task_type must be ADMIN or TECH")
|
|
if can_fix_self not in ("YES", "NO"):
|
|
raise ValueError(f"TechProblems_Solution row {idx}: can_fix_self must be YES or NO")
|
|
if need_result_feedback not in ("YES", "NO"):
|
|
raise ValueError(f"TechProblems_Solution row {idx}: need_result_feedback must be YES or NO")
|
|
seen_ids.add(problem_id)
|
|
solutions.append(
|
|
(
|
|
problem_id,
|
|
_require_non_empty(row[1], "TechProblems_Solution", idx, "problem_name"),
|
|
task_type,
|
|
can_fix_self,
|
|
need_result_feedback,
|
|
_require_non_empty(row[5], "TechProblems_Solution", idx, "solution_steps"),
|
|
(row[6] or "").strip(),
|
|
(row[7] or "").strip(),
|
|
)
|
|
)
|
|
return solutions
|
|
|
|
|
|
def _write_sync_data(pinpad_errors, terminal_data, tech_problems, tech_solutions):
|
|
instructions, keys, steps = terminal_data
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("BEGIN")
|
|
|
|
cur.execute("DELETE FROM terminal_instruction_steps")
|
|
cur.execute("DELETE FROM terminal_instruction_keys")
|
|
cur.execute("DELETE FROM terminal_instructions")
|
|
cur.execute("DELETE FROM pinpad_errors")
|
|
cur.execute("DELETE FROM tech_problem_solutions")
|
|
cur.execute("DELETE FROM tech_problems")
|
|
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO pinpad_errors (code, reason, action)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
pinpad_errors,
|
|
)
|
|
cur.executemany(
|
|
"INSERT INTO terminal_instructions (id, title) VALUES (?, ?)",
|
|
instructions,
|
|
)
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO terminal_instruction_keys
|
|
(instruction_id, key, key_type)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
keys,
|
|
)
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO terminal_instruction_steps
|
|
(instruction_id, step_order, type, content)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
steps,
|
|
)
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO tech_problems (id, task_type, keywords)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
tech_problems,
|
|
)
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO tech_problem_solutions
|
|
(problem_id, problem_name, task_type, can_fix_self,
|
|
need_result_feedback, solution_steps, tools_needed,
|
|
when_stop_and_report)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
tech_solutions,
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def sync_pinpad_errors(service):
|
|
rows = _fetch_rows(service, "PinPad!A2:C1000")
|
|
pinpad_errors = _validate_pinpad_errors(rows)
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM pinpad_errors")
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO pinpad_errors (code, reason, action)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
pinpad_errors,
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def sync_terminal(service):
|
|
terminal_data = _validate_terminal(
|
|
_fetch_rows(service, "Terminal_Instructions!A2:B1000"),
|
|
_fetch_rows(service, "Terminal_Keys!A2:C2000"),
|
|
_fetch_rows(service, "Terminal_Steps!A2:D3000"),
|
|
)
|
|
instructions, keys, steps = terminal_data
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM terminal_instruction_steps")
|
|
cur.execute("DELETE FROM terminal_instruction_keys")
|
|
cur.execute("DELETE FROM terminal_instructions")
|
|
cur.executemany(
|
|
"INSERT INTO terminal_instructions (id, title) VALUES (?, ?)",
|
|
instructions,
|
|
)
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO terminal_instruction_keys
|
|
(instruction_id, key, key_type)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
keys,
|
|
)
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO terminal_instruction_steps
|
|
(instruction_id, step_order, type, content)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
steps,
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def sync_tech_problems(service):
|
|
tech_problems, _problem_ids = _validate_tech_problems(
|
|
_fetch_rows(service, "TechProblems!A2:C2000")
|
|
)
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM tech_problem_solutions")
|
|
cur.execute("DELETE FROM tech_problems")
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO tech_problems (id, task_type, keywords)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
tech_problems,
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def sync_tech_problem_solutions(service):
|
|
tech_problems, problem_ids = _validate_tech_problems(
|
|
_fetch_rows(service, "TechProblems!A2:C2000")
|
|
)
|
|
tech_solutions = _validate_tech_solutions(
|
|
_fetch_rows(service, "TechProblems_Solution!A2:H2000"),
|
|
problem_ids,
|
|
)
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("DELETE FROM tech_problem_solutions")
|
|
cur.execute("DELETE FROM tech_problems")
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO tech_problems (id, task_type, keywords)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
tech_problems,
|
|
)
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO tech_problem_solutions
|
|
(problem_id, problem_name, task_type, can_fix_self,
|
|
need_result_feedback, solution_steps, tools_needed,
|
|
when_stop_and_report)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
tech_solutions,
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def sync_all():
|
|
service = get_google_service()
|
|
print("SYNC fetch data")
|
|
|
|
pinpad_errors = _validate_pinpad_errors(_fetch_rows(service, "PinPad!A2:C1000"))
|
|
terminal_data = _validate_terminal(
|
|
_fetch_rows(service, "Terminal_Instructions!A2:B1000"),
|
|
_fetch_rows(service, "Terminal_Keys!A2:C2000"),
|
|
_fetch_rows(service, "Terminal_Steps!A2:D3000"),
|
|
)
|
|
tech_problems, problem_ids = _validate_tech_problems(
|
|
_fetch_rows(service, "TechProblems!A2:C2000")
|
|
)
|
|
tech_solutions = _validate_tech_solutions(
|
|
_fetch_rows(service, "TechProblems_Solution!A2:H2000"),
|
|
problem_ids,
|
|
)
|
|
|
|
print("SYNC write data")
|
|
_write_sync_data(pinpad_errors, terminal_data, tech_problems, tech_solutions)
|
|
print("SYNC done")
|