456 lines
16 KiB
Python
Executable File
456 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Generate FETCH_TRANSACTIONS pool tasks from a live device screen.
|
||
|
||
Reads config.ini to find the DB, queries active materials, shows a menu
|
||
to pick which device+bank to scan, then dumps the transactions list from
|
||
the real device, taps into each detail, and generates pool JSON files.
|
||
|
||
Run from project root while the transactions list is open on the device:
|
||
|
||
python tests/mock_server/gen_fetch_transactions.py
|
||
|
||
Or specify config and serial:
|
||
|
||
python tests/mock_server/gen_fetch_transactions.py --config config.ini -s 148697059H000712
|
||
|
||
On Windows:
|
||
|
||
python tests\\mock_server\\gen_fetch_transactions.py
|
||
"""
|
||
|
||
import argparse
|
||
import configparser
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).parent
|
||
POOL_DIR = HERE / "scenarios" / "pool"
|
||
BOUNDS_RE = re.compile(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]")
|
||
|
||
# ---------------------------------------------------------------- config/db
|
||
|
||
def resolve_db_path(config_path: Path) -> Path:
|
||
config = configparser.ConfigParser()
|
||
config.read(str(config_path), encoding="utf-8")
|
||
raw = config.get("db", "dbPath", fallback="")
|
||
if not raw:
|
||
print(f"ERROR: [db] dbPath not found in {config_path}", file=sys.stderr)
|
||
sys.exit(1)
|
||
expanded = os.path.expanduser(raw)
|
||
p = Path(expanded)
|
||
if not p.is_absolute():
|
||
p = config_path.parent / p
|
||
return p.resolve()
|
||
|
||
|
||
def find_config(hint):
|
||
if hint:
|
||
p = Path(hint)
|
||
if p.exists():
|
||
return p
|
||
print(f"ERROR: config not found at {p}", file=sys.stderr)
|
||
sys.exit(1)
|
||
for c in [Path("config.ini"), Path("build/Release/config.ini"),
|
||
Path("build/config.ini"), Path("build/Debug/config.ini")]:
|
||
if c.exists():
|
||
return c.resolve()
|
||
print("ERROR: config.ini not found. Run from project root or use --config.", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
def query_active_materials(db_path: Path):
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.row_factory = sqlite3.Row
|
||
rows = conn.execute("""
|
||
SELECT d.id AS device_id, d.api_id, d.name AS device_name,
|
||
d.adb_serial,
|
||
bp.code AS bank_name, bp.bank_profile_id, bp.full_name,
|
||
m.material_id, m.last_numbers, m.card_number
|
||
FROM devices d
|
||
JOIN bank_profiles bp ON bp.device_id = d.id
|
||
JOIN materials m ON m.device_id = d.id AND m.app_code = bp.code
|
||
WHERE bp.status = 'active' AND m.status = 'active'
|
||
AND m.material_id != '' AND bp.bank_profile_id != ''
|
||
ORDER BY d.name, bp.code
|
||
""").fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
# ---------------------------------------------------------------- ADB helpers
|
||
|
||
def adb(serial, *args):
|
||
cmd = ["adb"]
|
||
if serial:
|
||
cmd += ["-s", serial]
|
||
cmd += list(args)
|
||
return subprocess.run(cmd, check=True, capture_output=True)
|
||
|
||
|
||
PACKAGE_TO_BANK = {
|
||
"ru.ozon.fintech.finance": "ozon",
|
||
"tj.dc.next1": "dushanbe_city_bank",
|
||
"app.blackwallet.ar": "black",
|
||
}
|
||
|
||
|
||
def get_foreground_package(serial) -> str | None:
|
||
"""Return the package name of the currently focused app."""
|
||
try:
|
||
r = adb(serial, "shell", "dumpsys", "window", "windows")
|
||
for line in r.stdout.decode("utf-8", errors="replace").splitlines():
|
||
if "mCurrentFocus" in line or "mFocusedApp" in line:
|
||
# e.g. mCurrentFocus=Window{... u0 tj.dc.next1/tj.dc.next1.MainActivity}
|
||
m = re.search(r"u\d+\s+(\S+)/", line)
|
||
if m:
|
||
return m.group(1)
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def dump_xml(serial) -> str:
|
||
adb(serial, "shell", "uiautomator", "dump", "/sdcard/_gen_ft.xml")
|
||
r = adb(serial, "exec-out", "cat", "/sdcard/_gen_ft.xml")
|
||
return r.stdout.decode("utf-8", errors="replace")
|
||
|
||
|
||
def get_device_tz(serial) -> timezone:
|
||
r = adb(serial, "shell", "date", "+%z")
|
||
s = r.stdout.decode().strip()
|
||
try:
|
||
sign = 1 if s[0] == "+" else -1
|
||
return timezone(timedelta(hours=sign * int(s[1:3]), minutes=sign * int(s[3:5])))
|
||
except (ValueError, IndexError):
|
||
print(f" warning: can't parse tz '{s}', assuming UTC+7")
|
||
return timezone(timedelta(hours=7))
|
||
|
||
|
||
def get_adb_serials():
|
||
r = subprocess.run(["adb", "devices"], check=True, capture_output=True)
|
||
serials = []
|
||
for line in r.stdout.decode().splitlines()[1:]:
|
||
parts = line.split()
|
||
if len(parts) >= 2 and parts[1] == "device":
|
||
serials.append(parts[0])
|
||
return serials
|
||
|
||
|
||
def tap(serial, x, y):
|
||
adb(serial, "shell", "input", "tap", str(x), str(y))
|
||
|
||
|
||
def back(serial):
|
||
adb(serial, "shell", "input", "keyevent", "KEYCODE_BACK")
|
||
|
||
|
||
# ---------------------------------------------------------------- Ozon
|
||
|
||
def parse_ozon_rows(xml):
|
||
rows = []
|
||
for m in re.finditer(
|
||
r'<node\s[^>]*?text="([^"]*(?:Пополнени|Переводы)[^"]*)"'
|
||
r'[^>]*?bounds="(\[\d+,\d+\]\[\d+,\d+\])"', xml):
|
||
text, bounds = m.group(1), m.group(2)
|
||
bm = BOUNDS_RE.match(bounds)
|
||
if not bm:
|
||
continue
|
||
x1, y1, x2, y2 = map(int, bm.groups())
|
||
if (y2 - y1) < 50 or "Расход" in text or "Доход" in text:
|
||
continue
|
||
am = re.search(r"([+\-−])\s*(\d[\d\s\u202f]*)", text)
|
||
if not am:
|
||
continue
|
||
sign = -1 if am.group(1) in ("-", "−") else 1
|
||
amount = sign * int(re.sub(r"[\s\u202f]", "", am.group(2)))
|
||
rows.append({"text": text, "cx": (x1+x2)//2, "cy": (y1+y2)//2, "amount": amount})
|
||
rows.sort(key=lambda r: r["cy"])
|
||
return rows
|
||
|
||
|
||
def is_ozon_list(xml):
|
||
return "Операции" in xml and ("Пополнени" in xml or "Переводы" in xml)
|
||
|
||
|
||
def extract_ozon_detail_time(xml):
|
||
for m in re.finditer(r'text="([^"]+)"', xml):
|
||
dm = re.match(r"(\d{2}\.\d{2}\.\d{4}),?\s+(\d{2}:\d{2})", m.group(1))
|
||
if dm:
|
||
return dm.group(1), dm.group(2) + ":00"
|
||
return None, None
|
||
|
||
|
||
# ---------------------------------------------------------------- Dushanbe
|
||
|
||
def parse_dushanbe_rows(xml):
|
||
rows = []
|
||
for m in re.finditer(
|
||
r'<node\s[^>]*?content-desc="([^"]*часть P2P/операции[^"]*)"'
|
||
r'[^>]*?bounds="(\[(\d+),(\d+)\]\[(\d+),(\d+)\])"', xml):
|
||
cd = m.group(1)
|
||
x1, y1, x2, y2 = int(m.group(3)), int(m.group(4)), int(m.group(5)), int(m.group(6))
|
||
h = y2 - y1
|
||
if h < 80 or h > 400:
|
||
continue
|
||
parts = cd.split("\n") if "\n" in cd else cd.split(" ")
|
||
amount, time_str = None, None
|
||
for p in parts:
|
||
p = p.strip()
|
||
try:
|
||
amount = int(float(p))
|
||
except ValueError:
|
||
pass
|
||
if re.match(r"^\s*\d{1,2}:\d{2}(:\d{2})?\s*$", p):
|
||
time_str = p.strip()
|
||
rows.append({"cx": (x1+x2)//2, "cy": (y1+y2)//2, "amount": amount, "time": time_str})
|
||
rows.sort(key=lambda r: r["cy"])
|
||
return rows
|
||
|
||
|
||
def is_dushanbe_list(xml):
|
||
return "Операции" in xml and "часть P2P" in xml
|
||
|
||
|
||
def extract_dushanbe_detail(xml):
|
||
m = re.search(r'content-desc="([^"]{30,})"', xml)
|
||
if not m:
|
||
return None, None, None
|
||
parts = [x.strip() for x in m.group(1).split(" ")]
|
||
info = {}
|
||
for i, x in enumerate(parts):
|
||
if x.endswith(":") and i + 1 < len(parts):
|
||
info[x] = parts[i + 1]
|
||
return info.get("Сумма:"), info.get("Дата:"), info.get("Время:")
|
||
|
||
|
||
# ---------------------------------------------------------------- Main
|
||
|
||
def short_label(amount, time_str):
|
||
sign = "plus" if amount >= 0 else "minus"
|
||
t = time_str.replace(":", "") if time_str else "unk"
|
||
return f"{sign}_{abs(amount)}_{t}"
|
||
|
||
|
||
def collect_tasks(serial, bank, mid, device_tz, kopecks, prefix):
|
||
xml = dump_xml(serial)
|
||
# Detect bank by foreground app package first, fall back to XML heuristics
|
||
fg_pkg = get_foreground_package(serial)
|
||
fg_bank = PACKAGE_TO_BANK.get(fg_pkg) if fg_pkg else None
|
||
if fg_bank:
|
||
print(f"Foreground app: {fg_pkg} -> {fg_bank}")
|
||
is_dush = fg_bank == "dushanbe_city_bank" or "dushanbe" in bank or is_dushanbe_list(xml)
|
||
is_ozon = not is_dush and (fg_bank == "ozon" or bank == "ozon" or is_ozon_list(xml))
|
||
|
||
if not is_ozon and not is_dush:
|
||
print("ERROR: not on a recognized transactions list", file=sys.stderr)
|
||
return []
|
||
|
||
tasks = []
|
||
seen = set()
|
||
|
||
if is_ozon:
|
||
print("Detected: Ozon")
|
||
baseline = parse_ozon_rows(xml)
|
||
print(f"rows: {len(baseline)}")
|
||
for i, r in enumerate(baseline, 1):
|
||
print(f" {i}. ({r['cx']},{r['cy']}) {r['amount']:+d} ₽ | {r['text'][:50]}")
|
||
|
||
for idx in range(1, len(baseline) + 1):
|
||
xml = dump_xml(serial)
|
||
if not is_ozon_list(xml):
|
||
print(f"[{idx}] left list, stop"); break
|
||
rows = parse_ozon_rows(xml)
|
||
if idx > len(rows):
|
||
break
|
||
row = rows[idx - 1]
|
||
print(f"\n[{idx}] tap ({row['cx']},{row['cy']}) {row['amount']:+d}")
|
||
tap(serial, row["cx"], row["cy"])
|
||
time.sleep(2.5)
|
||
d_xml = dump_xml(serial)
|
||
date_s, time_s = extract_ozon_detail_time(d_xml)
|
||
print(f" {date_s} {time_s}")
|
||
back(serial); time.sleep(1.3)
|
||
if not date_s or not time_s:
|
||
print(" SKIP"); continue
|
||
dt = datetime.strptime(f"{date_s} {time_s}", "%d.%m.%Y %H:%M:%S").replace(tzinfo=device_tz)
|
||
cat = dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
label = short_label(row["amount"], time_s)
|
||
while label in seen: label += "_2"
|
||
seen.add(label)
|
||
tasks.append({"bank_name": bank, "material_id": mid, "type": "FETCH_TRANSACTIONS",
|
||
"body": {"amount": row["amount"] * kopecks, "created_at": cat},
|
||
"_tag": f"{prefix}_{idx:02d}_{label}"})
|
||
|
||
elif is_dush:
|
||
print("Detected: Dushanbe")
|
||
baseline = parse_dushanbe_rows(xml)
|
||
print(f"rows: {len(baseline)}")
|
||
for i, r in enumerate(baseline, 1):
|
||
print(f" {i}. ({r['cx']},{r['cy']}) amt={r['amount']} time={r['time']}")
|
||
|
||
for idx in range(1, len(baseline) + 1):
|
||
xml = dump_xml(serial)
|
||
if not is_dushanbe_list(xml):
|
||
print(f"[{idx}] left list, stop"); break
|
||
rows = parse_dushanbe_rows(xml)
|
||
if idx > len(rows):
|
||
break
|
||
row = rows[idx - 1]
|
||
print(f"\n[{idx}] tap ({row['cx']},{row['cy']}) amt={row['amount']}")
|
||
tap(serial, row["cx"], row["cy"])
|
||
time.sleep(2.5)
|
||
d_xml = dump_xml(serial)
|
||
amt_s, date_s, time_s = extract_dushanbe_detail(d_xml)
|
||
print(f" amt={amt_s} {date_s} {time_s}")
|
||
back(serial); time.sleep(1.3)
|
||
if not date_s or not time_s:
|
||
print(" SKIP"); continue
|
||
amount = int(float(amt_s)) if amt_s else (row["amount"] or 0)
|
||
fmt = "%d.%m.%Y %H:%M:%S" if len(time_s) > 5 else "%d.%m.%Y %H:%M"
|
||
dt = datetime.strptime(f"{date_s} {time_s}", fmt).replace(tzinfo=device_tz)
|
||
cat = dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
label = short_label(amount, time_s)
|
||
while label in seen: label += "_2"
|
||
seen.add(label)
|
||
tasks.append({"bank_name": bank, "material_id": mid, "type": "FETCH_TRANSACTIONS",
|
||
"body": {"amount": amount * kopecks, "created_at": cat},
|
||
"_tag": f"{prefix}_{idx:02d}_{label}"})
|
||
|
||
return tasks
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Generate FETCH_TRANSACTIONS pool from live device")
|
||
parser.add_argument("--config", "-c", default=None, help="Path to config.ini")
|
||
parser.add_argument("--serial", "-s", default=None, help="ADB serial (prompted if multiple)")
|
||
parser.add_argument("--kopecks", type=int, default=100, help="Amount multiplier (default 100)")
|
||
parser.add_argument("--mock-host", default="127.0.0.1")
|
||
parser.add_argument("--mock-port", type=int, default=8888)
|
||
args = parser.parse_args()
|
||
|
||
# Find config and DB
|
||
for c in ([Path(args.config)] if args.config else
|
||
[Path("config.ini"), Path("build/Release/config.ini"),
|
||
Path("build/config.ini")]):
|
||
if c.exists():
|
||
config_path = c.resolve(); break
|
||
else:
|
||
print("ERROR: config.ini not found", file=sys.stderr); return 1
|
||
|
||
db_path = resolve_db_path(config_path)
|
||
print(f"config: {config_path}")
|
||
print(f"db: {db_path}")
|
||
|
||
materials = query_active_materials(db_path)
|
||
if not materials:
|
||
print("No active materials in DB"); return 1
|
||
|
||
# Find connected ADB devices
|
||
serials = get_adb_serials()
|
||
if not serials:
|
||
print("ERROR: no ADB devices connected"); return 1
|
||
|
||
# Match DB devices to connected ADB serials
|
||
options = []
|
||
for mat in materials:
|
||
adb_serial = mat["adb_serial"]
|
||
if adb_serial not in serials:
|
||
continue
|
||
dev_label = re.sub(r"-[a-f0-9]{10,}$", "", mat["device_name"])
|
||
options.append({**mat, "serial": adb_serial, "label": dev_label})
|
||
|
||
if not options:
|
||
print("No active materials match connected ADB devices.")
|
||
print(f" DB serials: {[m['adb_serial'] for m in materials]}")
|
||
print(f" ADB serials: {serials}")
|
||
return 1
|
||
|
||
# Menu
|
||
print(f"\nAvailable materials ({len(options)}):")
|
||
for i, o in enumerate(options, 1):
|
||
print(f" {i}. {o['label']} / {o['bank_name']} / {o['full_name'] or '?'} "
|
||
f"(material {o['material_id'][:8]}… serial={o['serial']})")
|
||
|
||
if len(options) == 1:
|
||
choice = 0
|
||
print(f"\n→ auto-selected: {options[0]['label']} / {options[0]['bank_name']}")
|
||
else:
|
||
try:
|
||
inp = input(f"\nPick [1-{len(options)}] or 'all': ").strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
return 0
|
||
if inp.lower() == "all":
|
||
choice = -1
|
||
else:
|
||
choice = int(inp) - 1
|
||
if choice < 0 or choice >= len(options):
|
||
print("Invalid choice"); return 1
|
||
|
||
to_scan = options if choice == -1 else [options[choice]]
|
||
all_tasks = []
|
||
|
||
for o in to_scan:
|
||
serial = args.serial or o["serial"]
|
||
bank = o["bank_name"]
|
||
mid = o["material_id"]
|
||
prefix = f"{o['label'].lower().replace(' ', '_')}_{bank}"
|
||
device_tz = get_device_tz(serial)
|
||
print(f"\n{'='*60}")
|
||
print(f"Scanning: {o['label']} / {bank} ({mid[:8]}…)")
|
||
print(f"Serial: {serial} TZ: UTC{device_tz.utcoffset(None)}")
|
||
print(f"{'='*60}")
|
||
|
||
input(f"\nOpen {bank} transactions list on {o['label']} and press Enter...")
|
||
|
||
tasks = collect_tasks(serial, bank, mid, device_tz, args.kopecks, prefix)
|
||
all_tasks.extend(tasks)
|
||
|
||
if not all_tasks:
|
||
print("\nNo tasks generated"); return 1
|
||
|
||
# Write pool
|
||
POOL_DIR.mkdir(parents=True, exist_ok=True)
|
||
# Remove old files for scanned prefixes
|
||
scanned_prefixes = set()
|
||
for o in to_scan:
|
||
scanned_prefixes.add(f"{o['label'].lower().replace(' ', '_')}_{o['bank_name']}")
|
||
for old in POOL_DIR.glob("*.json"):
|
||
for pfx in scanned_prefixes:
|
||
if old.name.startswith(pfx + "_ft_"):
|
||
old.unlink(); break
|
||
|
||
for i, t in enumerate(all_tasks, 1):
|
||
tag = t["_tag"]
|
||
fname = f"{tag.rsplit('_', 1)[0]}_ft_{i:02d}.json"
|
||
(POOL_DIR / fname).write_text(json.dumps(t, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
||
|
||
print(f"\n{len(all_tasks)} tasks written to {POOL_DIR}/")
|
||
for t in all_tasks:
|
||
b = t["body"]
|
||
print(f" {t['_tag']}: amount={b['amount']:+d} created_at={b['created_at']}")
|
||
|
||
# Inject into mock if running
|
||
try:
|
||
import urllib.request
|
||
url = f"http://{args.mock_host}:{args.mock_port}/_admin/inject"
|
||
req = urllib.request.Request(url, data=json.dumps(all_tasks).encode(),
|
||
headers={"Content-Type": "application/json"}, method="POST")
|
||
with urllib.request.urlopen(req, timeout=3) as resp:
|
||
print(f"\nInjected {len(all_tasks)} tasks into mock -> {resp.status}")
|
||
except Exception as e:
|
||
print(f"\nMock not reachable ({e}) — tasks saved to files only")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|