felica_suica.py

"""
Suica/PASMO 历史块解析(64 位直驱用)。
输入: read_service 读到的 20 个 16 字节块(newest->oldest), 卡 IDm。
输出: 和现有 32 位路一致的记录 dict, 直接喂 history_store / 地图。

站名: サイバネコード(地区/線区/駅順)-> 站名, 用公开 StationCode.csv(运行时下载缓存)。
块格式参考: Metrodroid / nfcpy。日期/余额/连番已用真卡验证一致。
"""
from __future__ import annotations
import csv
import datetime
import os
import urllib.request

try:
    import paths  # 主程序里复用可写目录
    _CACHE_DIR = paths.app_data_dir()
except Exception:
    _CACHE_DIR = os.path.dirname(os.path.abspath(__file__))  # 独立跑时放脚本目录

_CSV_URL = "https://cdn.jsdelivr.net/gh/m2wasabi/nfcpy-suica-sample@master/StationCode.csv"
_CSV_CACHE = os.path.join(_CACHE_DIR, "suica_station_code.csv")

# 端末种别 -> 类型(简化; 参考 Metrodroid)
_RAIL_TERMINALS = {0x03, 0x04, 0x05, 0x07, 0x08, 0x12, 0x14, 0x15, 0x16, 0x17, 0x18,
                   0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1f, 0x46, 0x48}
_BUS_TERMINALS = {0x05, 0x0d, 0x66, 0x67, 0x68, 0x69}
# 处理种别
_PROC_CHARGE = {0x02, 0x06, 0x0a}          # チャージ/入金
_PROC_PURCHASE = {0x46, 0x49, 0x73, 0xc6}  # 物販等


def _ensure_csv() -> str | None:
    if os.path.exists(_CSV_CACHE) and os.path.getsize(_CSV_CACHE) > 0:
        return _CSV_CACHE
    try:
        req = urllib.request.Request(_CSV_URL, headers={"User-Agent": "NekoCardReader/1.0"})
        with urllib.request.urlopen(req, timeout=30) as r:
            data = r.read()
        with open(_CSV_CACHE, "wb") as f:
            f.write(data)
        return _CSV_CACHE
    except Exception:
        return None


_STATIONS = None          # (area,line,station) -> (company, line_name, station_name)
_BY_LINE_ST = None        # (line,station) -> (company, line_name, station_name)  兜底(忽略 area)


def _load_stations():
    global _STATIONS, _BY_LINE_ST
    if _STATIONS is not None:
        return
    _STATIONS, _BY_LINE_ST = {}, {}
    path = _ensure_csv()
    if not path:
        return
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for row in csv.reader(f):
                if len(row) < 6:
                    continue
                try:
                    area, line, st = int(row[0]), int(row[1]), int(row[2])
                except ValueError:
                    continue
                rec = (row[3], row[4], row[5])
                _STATIONS[(area, line, st)] = rec
                _BY_LINE_ST.setdefault((line, st), rec)
    except OSError:
        pass


def _station(area, line, st):
    """返回 (公司, 线名, 站名) 或 None。area 不确定时退化到 (line,station)。"""
    if line == 0 and st == 0:
        return None
    _load_stations()
    return _STATIONS.get((area, line, st)) or (_BY_LINE_ST or {}).get((line, st))


def parse_blocks(raw: bytes, idm: str = "") -> list[dict]:
    """raw = 拼接的 16*N 字节块(newest->oldest)。返回记录列表。"""
    n = len(raw) // 16
    rows = []
    prev_bal = None
    for i in range(n):
        b = raw[i*16:i*16+16]
        if b == b"\x00" * 16:
            continue
        term, proc = b[0], b[1]
        draw = (b[4] << 8) | b[5]
        year = 2000 + ((draw >> 9) & 0x7f)
        month = (draw >> 5) & 0xf
        day = draw & 0x1f
        bal = b[10] | (b[11] << 8)
        seq = (b[12] << 16) | (b[13] << 8) | b[14]
        region = b[15]
        in_st = _station(region, b[6], b[7])
        out_st = _station(region, b[8], b[9])

        is_charge = proc in _PROC_CHARGE
        if term in _BUS_TERMINALS and not in_st and not out_st:
            memo = "バス/路面等"
        elif is_charge:
            memo = "チャージ"
        elif proc in _PROC_PURCHASE:
            memo = "物販"
        else:
            memo = ""

        try:
            date = datetime.date(year, month, day) if month and day else None
        except ValueError:
            date = None
        rows.append({
            "date": f"{year}/{month:02d}/{day:02d}" if month and day else "",
            "in_company": in_st[0] if in_st else "",
            "in_station": in_st[2] if in_st else "",
            "out_company": out_st[0] if out_st else "",
            "out_station": out_st[2] if out_st else "",
            "memo": memo,
            "balance": bal,
            "expense": 0,            # 下面两遍法补
            "in_commute": False,
            "out_commute": False,
            "log_id": seq,
            "_term": term, "_proc": proc,
        })
    # expense: 本条交易额 = 上一条(更旧)余额 - 本条余额。正=消费, 负=入金。
    # rows 是 newest->oldest, rows[k+1] 即更旧一条。
    for k in range(len(rows)):
        if k + 1 < len(rows):
            rows[k]["expense"] = rows[k + 1]["balance"] - rows[k]["balance"]
    return rows


if __name__ == "__main__":
    # 离线自测: 真卡块
    test = bytes.fromhex(
        "050d000f34cf0c5000006b000000e300"
        "1601000434cde33de35e65010000e200"
        "1601000234cde32de33d17020000e000"
        "1b023f0034cd9b8100006b040000db00")
    for r in parse_blocks(test, "013972100107e6f5"):
        print(r["log_id"], r["date"], r["balance"], "円",
              f"{r['in_station']}->{r['out_station']}" if r["in_station"] else r["memo"],
              "exp=", r["expense"])