felica_reader.py

"""
FeliCa 读卡后端 — 通过 PowerShell 桥接调用 Sony SFCAccLib.dll
完全使用 Sony 原装驱动，无需 Zadig / libusb

架构:
  Python (64-bit GUI)
    └─ subprocess ──► 32-bit PowerShell
                         └─ .NET 反射 ──► SFCV.exe (SFCardViewer.FeliCaIO)
                                              └─ P/Invoke ──► SFCAccLib.dll
                                                                 └─ Sony RC-S330 驱动
"""

from __future__ import annotations
import json
import os
import subprocess
import sys
import datetime
import threading
from dataclasses import dataclass, field
from typing import Optional

# 路径配置
import paths
_HERE = os.path.dirname(os.path.abspath(__file__))
_NATIVE_PS1 = paths.resource_path("felica_native.ps1")  # 纯 P/Invoke 直驱(不依赖 SFCV.exe)
_PS32 = r"C:\Windows\SysWOW64\WindowsPowerShell\v1.0\powershell.exe"
# 含 SFCAccLib.dll/SFCRep.dll(+felica.dll) 的目录候选(外部安装兜底; 拼写有笔误版)
_DLL_DIR_CANDIDATES = [
    r"C:\Program Files (x86)\Sony Corporation\SFCard Viewer 2",
    r"C:\Program Files (x86)\Sony Coproration\SFCard Viewer 2",
    r"C:\Program Files\Sony Corporation\SFCard Viewer 2",
    r"C:\Program Files\Sony Coproration\SFCard Viewer 2",
]


def resolve_dll_dir(hint: str = "") -> str:
    """
    返回含 SFCAccLib.dll 的目录。优先级:
      用户设置(目录或 SFCV.exe 路径) → 自带 sfcv_runtime → 外部安装。
    felica.dll 由 felica_native.ps1 自行在 Common Files 里兜底查找。
    """
    if hint:
        d = hint if os.path.isdir(hint) else os.path.dirname(hint)
        if d and os.path.exists(os.path.join(d, "SFCAccLib.dll")):
            return d
    bundled = paths.resource_path("sfcv_runtime")
    if os.path.exists(os.path.join(bundled, "SFCAccLib.dll")):
        return bundled
    for d in _DLL_DIR_CANDIDATES:
        if os.path.exists(os.path.join(d, "SFCAccLib.dll")):
            return d
    return bundled


# ── 数据结构 ──────────────────────────────────────────────────────────────────
@dataclass
class Transaction:
    index: int
    date: Optional[datetime.date]
    date_str: str
    in_station: str
    in_company: str
    out_station: str
    out_company: str
    memo: str
    balance: int        # 交易后余额 (円)
    amount: int         # 变动金额 (负=消费, 正=入金)
    in_commute: bool
    out_commute: bool
    log_id: int
    # 统一字段(中国卡用; 日本卡留默认)
    country: str = "日本"          # 日本 / 中国
    region: str = ""               # 地区(日本地方 / 中国大区); 空=按坐标推断
    city: str = ""                 # 城市(中国)
    line: str = ""                 # 线路(中国)
    in_lat: Optional[float] = None
    in_lng: Optional[float] = None
    out_lat: Optional[float] = None
    out_lng: Optional[float] = None

    @property
    def terminal_name(self) -> str:
        """根据 memo / in_station 推断交易类型"""
        if self.memo:
            return self.memo
        if self.amount > 0:
            return "チャージ"
        if not self.in_station and not self.out_station:
            return "物販/その他"
        return "電車"

    @property
    def process_name(self) -> str:
        if self.amount > 0:
            return "入金"
        if self.out_station:
            return "SF出場"
        return "SF利用"

    def to_dict(self) -> dict:
        return {
            "index": self.index,
            "date": self.date.isoformat() if self.date else None,
            "date_str": self.date_str,
            "in_station": self.in_station,
            "in_company": self.in_company,
            "out_station": self.out_station,
            "out_company": self.out_company,
            "memo": self.memo,
            "balance": self.balance,
            "amount": self.amount,
            "in_commute": self.in_commute,
            "out_commute": self.out_commute,
            "log_id": self.log_id,
            "country": self.country,
            "region": self.region,
            "city": self.city,
            "line": self.line,
            "in_lat": self.in_lat, "in_lng": self.in_lng,
            "out_lat": self.out_lat, "out_lng": self.out_lng,
        }


@dataclass
class CardData:
    balance: int
    transactions: list[Transaction]
    read_at: datetime.datetime = field(default_factory=datetime.datetime.now)
    idm: str = ""

    def to_dict(self) -> dict:
        return {
            "balance": self.balance,
            "read_at": self.read_at.isoformat(),
            "transactions": [t.to_dict() for t in self.transactions],
        }


# ── 解析 PowerShell 输出 ─────────────────────────────────────────────────────
def _parse_date(s: str) -> Optional[datetime.date]:
    if not s:
        return None
    for fmt in ("%Y/%m/%d", "%Y-%m-%d", "%m/%d/%Y",
                "%Y年%m月%d日", "%Y.%m.%d"):
        try:
            return datetime.datetime.strptime(s.strip(), fmt).date()
        except ValueError:
            pass
    return None


def _parse_json_result(raw_json: str | dict) -> CardData:
    # 兼容旧桥接层的 JSON 字符串，也接受进程内后端直接返回的 dict。
    data = raw_json if isinstance(raw_json, dict) else json.loads(raw_json)

    if data.get("error"):
        raise RuntimeError(data["error"])

    balance = int(data.get("balance", 0))
    history_raw = data.get("history", [])

    transactions: list[Transaction] = []
    for i, h in enumerate(history_raw):
        date_str = str(h.get("date", "") or "")
        t = Transaction(
            index=i,
            date=_parse_date(date_str),
            date_str=date_str,
            in_station=str(h.get("in_station", "") or ""),
            in_company=str(h.get("in_company", "") or ""),
            out_station=str(h.get("out_station", "") or ""),
            out_company=str(h.get("out_company", "") or ""),
            memo=str(h.get("memo", "") or ""),
            balance=int(h.get("balance", 0)),
            # 卡内 expense 字段: 正=消费(扣款), 负=充值(入金)。
            # amount 统一为「余额变化」: 正=入金, 负=消费。
            amount=-int(h.get("expense", 0)),
            in_commute=bool(h.get("in_commute", False)),
            out_commute=bool(h.get("out_commute", False)),
            log_id=int(h.get("log_id", 0)),
        )
        transactions.append(t)

    idm = str(data.get("idm", "") or "")
    return CardData(balance=balance, transactions=transactions, idm=idm)


# ── 直驱读卡(全 64 位进程内, 纯 ctypes 调 felica.dll, 不要 SFCV/SFCAccLib/子进程)──
def read_card_via_bridge(timeout_sec: int = 15,
                         sfcv_path: str = "",
                         on_waiting=None) -> CardData:
    """
    进程内 64 位读 Suica/PASMO(felica64_reader)。返回 CardData, 失败抛异常。
    sfcv_path 仅为兼容旧签名, 不再使用。
    """
    import felica64_reader
    if on_waiting:
        on_waiting()
    r = felica64_reader.read_suica(timeout_sec=timeout_sec)
    if r.get("error"):
        raise RuntimeError(r["error"])
    return _parse_json_result(r)


# ── Mock 数据（无读卡器时测试用）────────────────────────────────────────────
def _make_mock_data() -> CardData:
    import random
    random.seed(42)
    transactions = []
    balance = 3240

    station_pairs = [
        ("新宿", "JR東日本", "渋谷", "JR東日本"),
        ("渋谷", "JR東日本", "品川", "JR東日本"),
        ("品川", "JR東日本", "新宿", "JR東日本"),
        ("新宿", "東京Metro", "表参道", "東京Metro"),
        ("表参道", "東京Metro", "新宿", "東京Metro"),
        ("上野", "JR東日本", "東京", "JR東日本"),
        ("", "", "", ""),  # 物販
    ]

    base = datetime.date(2024, 6, 14)
    for i in range(20):
        days_back = 20 - i
        date = base - datetime.timedelta(days=days_back // 2)
        sp = station_pairs[i % len(station_pairs)]
        if sp[0]:
            fare = random.choice([140, 176, 210, 308, 420])
            balance -= fare
            amount = -fare
            memo = ""
        else:
            # 物販
            spend = random.choice([150, 220, 480])
            balance -= spend
            amount = -spend
            memo = "コンビニ"

        # 時々チャージ
        if random.random() < 0.25:
            charge = random.choice([1000, 2000, 3000])
            balance += charge
            transactions.append(Transaction(
                index=i, date=date, date_str=date.strftime("%Y/%m/%d"),
                in_station="", in_company="", out_station="", out_company="",
                memo="オートチャージ", balance=balance, amount=charge,
                in_commute=False, out_commute=False, log_id=i*100+1,
            ))
            continue

        transactions.append(Transaction(
            index=i, date=date, date_str=date.strftime("%Y/%m/%d"),
            in_station=sp[0], in_company=sp[1],
            out_station=sp[2], out_company=sp[3],
            memo=memo, balance=max(0, balance), amount=amount,
            in_commute=False, out_commute=False, log_id=i*100,
        ))

    return CardData(balance=max(0, balance), transactions=transactions, idm="DEMO")


# ── 异步读卡封装（供 GUI 调用）───────────────────────────────────────────────
class AsyncCardReader:
    def __init__(self, demo=False, sfcv_path="", timeout=15):
        self.demo = demo
        self.sfcv_path = sfcv_path
        self.timeout = timeout
        self._thread: Optional[threading.Thread] = None
        self._stop = threading.Event()

    def start(self, on_card, on_error, on_waiting=None):
        self._stop.clear()

        def run():
            try:
                if self.demo:
                    import time; time.sleep(1.0)
                    on_card(_make_mock_data())
                    return
                data = read_card_via_bridge(
                    timeout_sec=self.timeout,
                    sfcv_path=self.sfcv_path,
                    on_waiting=on_waiting,
                )
                on_card(data)
            except Exception as e:
                on_error(e)

        self._thread = threading.Thread(target=run, daemon=True)
        self._thread.start()

    def stop(self):
        self._stop.set()


if __name__ == "__main__":
    # 快速自测
    if "--demo" in sys.argv:
        data = _make_mock_data()
        print(f"[DEMO] 余额: {data.balance}円  记录数: {len(data.transactions)}")
        for t in data.transactions[:5]:
            print(f"  [{t.date_str}] {t.in_station}→{t.out_station}  "
                  f"{t.amount:+d}円  余额:{t.balance}円")
    else:
        print("读取真实卡片…")
        try:
            data = read_card_via_bridge(timeout_sec=15)
            print(f"余额: {data.balance}円  记录数: {len(data.transactions)}")
            for t in data.transactions[:5]:
                print(f"  [{t.date_str}] {t.in_station}→{t.out_station}  "
                      f"{t.amount:+d}円  余额:{t.balance}円")
        except Exception as e:
            print(f"错误: {e}")