gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement a new contract-based payment system with wallet binding, payment intents, and subscription plan management.

Browse Source
This commit is contained in:
2569718930@qq.com
2026-03-14 10:35:30 +08:00
parent be61c39795
commit 02e35a3a5f
10 changed files with 698 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bot/runtime_coordinator.py
+46 -3
View File
@@ -6,7 +6,10 @@ from datetime import datetime
from importlib import import_module
from typing import Any, Callable, Dict, List, Optional
from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env
from src.utils.telegram_chat_ids import (
get_polymarket_wallet_activity_chat_ids_from_env,
get_telegram_chat_ids_from_env,
)
def _env_bool(name: str, default: bool) -> bool:
@@ -87,6 +90,7 @@ class StartupCoordinator:
self._start_polygon_wallet_loop(),
self._start_polymarket_wallet_activity_loop(),
self._start_weekly_reward_loop(),
self._start_payment_event_loop(),
self._start_payment_confirm_loop(),
]
self._runtime_status = RuntimeStatus(
@@ -210,7 +214,7 @@ class StartupCoordinator:
def _start_polymarket_wallet_activity_loop(self) -> LoopStatus:
enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False)
chat_ids = get_telegram_chat_ids_from_env()
chat_ids = get_polymarket_wallet_activity_chat_ids_from_env()
users_count = _parse_csv_count(os.getenv("POLYMARKET_WALLET_ACTIVITY_USERS"))
poll = max(5, _env_int("POLYMARKET_WALLET_ACTIVITY_INTERVAL_SEC", 20))
details = {
@@ -221,7 +225,9 @@ class StartupCoordinator:
}
validation_error = None
if not chat_ids:
validation_error = "missing_TELEGRAM_CHAT_IDS"
validation_error = (
"missing_POLYMARKET_WALLET_ACTIVITY_CHAT_IDS_or_TELEGRAM_CHAT_IDS"
)
elif users_count == 0:
validation_error = "missing_POLYMARKET_WALLET_ACTIVITY_USERS"
return self._start_with_validation(
@@ -300,6 +306,43 @@ class StartupCoordinator:
).start_payment_confirm_loop(),
)
def _start_payment_event_loop(self) -> LoopStatus:
enabled = _env_bool("POLYWEATHER_PAYMENT_EVENT_LOOP_ENABLED", True)
details = {
"interval_sec": max(
5, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_INTERVAL_SEC", 20)
),
"lookback_blocks": max(
500,
_env_int(
"POLYWEATHER_PAYMENT_EVENT_LOOP_START_LOOKBACK_BLOCKS",
5000,
),
),
"step_blocks": min(
49999,
max(100, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_STEP_BLOCKS", 2000)),
),
"max_events": max(
10, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_MAX_EVENTS_PER_CYCLE", 200)
),
"payment_enabled": _env_bool("POLYWEATHER_PAYMENT_ENABLED", False),
"chain_id": _env_int("POLYWEATHER_PAYMENT_CHAIN_ID", 137),
}
validation_error = None
if not bool(details["payment_enabled"]):
validation_error = "payment_service_disabled"
return self._start_with_validation(
key="payment_event",
label="支付事件监听",
configured_enabled=enabled,
details=details,
validation_error=validation_error,
starter=lambda: import_module(
"src.payments.event_loop"
).start_payment_event_loop(),
)
def render_runtime_status_html(status: RuntimeStatus) -> str:
lines = [
src/onchain/polymarket_wallet_activity_watcher.py
+8 -3
View File
@@ -9,7 +9,9 @@ from typing import Any, Dict, List, Optional, Tuple
import requests
from loguru import logger
from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env
from src.utils.telegram_chat_ids import (
get_polymarket_wallet_activity_chat_ids_from_env,
)
def _env_bool(name: str, default: bool) -> bool:
@@ -577,7 +579,7 @@ def _filter_changes_by_position_value(
def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread]:
enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False)
chat_ids = get_telegram_chat_ids_from_env()
chat_ids = get_polymarket_wallet_activity_chat_ids_from_env()
users = _parse_addresses(os.getenv("POLYMARKET_WALLET_ACTIVITY_USERS"))
user_aliases = _parse_address_aliases(
os.getenv("POLYMARKET_WALLET_ACTIVITY_USER_ALIASES")
@@ -591,7 +593,10 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread
logger.info("polymarket wallet activity watcher disabled")
return None
if not chat_ids:
logger.warning("polymarket wallet activity watcher skipped: TELEGRAM_CHAT_IDS is not set")
logger.warning(
"polymarket wallet activity watcher skipped: "
"POLYMARKET_WALLET_ACTIVITY_CHAT_IDS/CHAT_ID and TELEGRAM_CHAT_IDS/CHAT_ID are empty"
)
return None
if not users:
logger.warning("polymarket wallet activity watcher skipped: POLYMARKET_WALLET_ACTIVITY_USERS is empty")
src/payments/contract_checkout.py
+75
View File
@@ -103,6 +103,21 @@ def _normalize_address(address: Any) -> str:
return Web3.to_checksum_address(text).lower()
def _normalize_order_id_hex(order_id_hex: Any) -> str:
text = str(order_id_hex or "").strip().lower()
if not text:
return ""
if not text.startswith("0x"):
text = f"0x{text}"
if len(text) != 66:
return ""
try:
int(text[2:], 16)
except Exception:
return ""
return text
def _now_utc() -> datetime:
return datetime.now(timezone.utc)
@@ -1321,6 +1336,66 @@ class PaymentContractCheckoutService:
)
return out
def list_open_intents_by_order_id(
self,
order_id_hex: str,
limit: int = 10,
) -> List[Dict[str, Any]]:
"""
Find intents by on-chain order id for event-driven reconciliation.
Includes created/submitted intents; confirmed intents are returned too for idempotent skip.
"""
self._ensure_enabled()
normalized_order = _normalize_order_id_hex(order_id_hex)
if not normalized_order:
return []
safe_limit = max(1, min(int(limit or 10), 50))
rows = self._rest(
"GET",
"payment_intents",
params={
"select": (
"id,user_id,status,tx_hash,order_id_hex,plan_id,token_address,"
"amount_units,payment_mode,allowed_wallet,chain_id,created_at,updated_at"
),
"order_id_hex": f"eq.{normalized_order}",
"chain_id": f"eq.{self.chain_id}",
"status": "in.(created,submitted,confirmed)",
"order": "created_at.desc",
"limit": str(safe_limit),
},
allowed_status=[200],
)
if not isinstance(rows, list):
return []
out: List[Dict[str, Any]] = []
for row in rows:
if not isinstance(row, dict):
continue
intent_id = str(row.get("id") or "").strip()
user_id = str(row.get("user_id") or "").strip()
status = str(row.get("status") or "").strip().lower()
if not intent_id or not user_id or not status:
continue
out.append(
{
"intent_id": intent_id,
"user_id": user_id,
"status": status,
"tx_hash": str(row.get("tx_hash") or "").strip().lower(),
"order_id_hex": str(row.get("order_id_hex") or "").strip().lower(),
"plan_id": int(row.get("plan_id") or 0),
"token_address": _normalize_address(row.get("token_address")),
"amount_units": int(row.get("amount_units") or 0),
"payment_mode": str(row.get("payment_mode") or "").strip().lower(),
"allowed_wallet": _normalize_address(row.get("allowed_wallet")),
"created_at": row.get("created_at"),
"updated_at": row.get("updated_at"),
}
)
return out
def submit_intent_tx(
self,
user_id: str,
src/payments/event_loop.py
+416
View File
@@ -0,0 +1,416 @@
from __future__ import annotations
import json
import os
import threading
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from loguru import logger
from web3 import Web3
from src.payments import PAYMENT_CHECKOUT, PaymentCheckoutError
def _env_bool(name: str, default: bool) -> bool:
raw = os.getenv(name)
if raw is None:
return default
return raw.strip().lower() in {"1", "true", "yes", "on"}
def _env_int(name: str, default: int, min_value: int = 0) -> int:
raw = os.getenv(name)
if raw is None:
return default
try:
value = int(raw)
except Exception:
return default
return max(min_value, value)
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _normalize_address(address: Any) -> str:
text = str(address or "").strip()
if not text or not Web3.is_address(text):
return ""
return Web3.to_checksum_address(text).lower()
def _to_hex(value: Any) -> str:
try:
return str(Web3.to_hex(value or b"")).lower()
except Exception:
return ""
def _state_file() -> str:
custom = str(
os.getenv("POLYWEATHER_PAYMENT_EVENT_LOOP_STATE_PATH") or ""
).strip()
if custom:
return custom
runtime_dir = str(os.getenv("POLYWEATHER_RUNTIME_DATA_DIR") or "").strip()
if runtime_dir:
return os.path.join(runtime_dir, "payment_event_loop_state.json")
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
return os.path.join(root, "data", "payment_event_loop_state.json")
def _load_state(path: str) -> Dict[str, Any]:
if not os.path.exists(path):
return {}
try:
with open(path, "r", encoding="utf-8") as fh:
raw = json.load(fh)
return raw if isinstance(raw, dict) else {}
except Exception as exc:
logger.warning(f"payment event loop state load failed: {exc}")
return {}
def _save_state(path: str, state: Dict[str, Any]) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True)
tmp_path = f"{path}.tmp"
with open(tmp_path, "w", encoding="utf-8") as fh:
json.dump(state, fh, ensure_ascii=False, indent=2)
os.replace(tmp_path, path)
def _is_pending_confirm_error(exc: PaymentCheckoutError) -> bool:
detail = str(exc.detail or "").lower()
if exc.status_code in {404, 408, 502, 503}:
return True
if exc.status_code == 409 and (
"confirmations not enough" in detail or "tx indexed partially" in detail
):
return True
return False
def _is_nonfatal_submit_error(exc: PaymentCheckoutError) -> bool:
detail = str(exc.detail or "").lower()
if exc.status_code in {502, 503, 408}:
return True
if exc.status_code == 409 and (
"intent status is submitted" in detail
or "intent status is confirmed" in detail
or "cannot submit" in detail
):
return True
return False
def _decode_order_paid_log(log_item: Any) -> Optional[Dict[str, Any]]:
log_get = getattr(log_item, "get", None)
if not callable(log_get):
return None
address = _normalize_address(log_get("address"))
if not address:
return None
try:
contract = PAYMENT_CHECKOUT._get_contract(address) # noqa: SLF001
event_obj = contract.events.OrderPaid().process_log(log_item)
except Exception:
return None
event_get = getattr(event_obj, "get", None)
args = event_get("args") if callable(event_get) else getattr(event_obj, "args", None)
if not args:
return None
args_get = getattr(args, "get", None)
if not callable(args_get):
return None
order_id_hex = _to_hex(args_get("orderId"))
payer = _normalize_address(args_get("payer"))
token = _normalize_address(args_get("token"))
plan_id = int(args_get("planId") or 0)
amount_units = int(args_get("amount") or 0)
tx_hash = _to_hex(log_get("transactionHash"))
block_number = int(log_get("blockNumber") or 0)
log_index = int(log_get("logIndex") or 0)
if not (order_id_hex and payer and token and tx_hash and plan_id > 0 and amount_units > 0):
return None
return {
"order_id_hex": order_id_hex,
"payer": payer,
"plan_id": plan_id,
"token_address": token,
"amount_units": amount_units,
"receiver_contract": address,
"tx_hash": tx_hash,
"block_number": block_number,
"log_index": log_index,
}
def _event_key(event_row: Dict[str, Any]) -> str:
return f"{event_row.get('tx_hash')}:{int(event_row.get('log_index') or 0)}"
def _select_matching_intents(
intents: List[Dict[str, Any]],
event_row: Dict[str, Any],
) -> List[Dict[str, Any]]:
out: List[Dict[str, Any]] = []
for row in intents:
if not isinstance(row, dict):
continue
if int(row.get("plan_id") or 0) != int(event_row.get("plan_id") or 0):
continue
if _normalize_address(row.get("token_address")) != _normalize_address(
event_row.get("token_address")
):
continue
if int(row.get("amount_units") or 0) != int(event_row.get("amount_units") or 0):
continue
out.append(row)
if out:
return out
return intents
def _runner() -> None:
enabled = _env_bool("POLYWEATHER_PAYMENT_EVENT_LOOP_ENABLED", True)
if not enabled:
logger.info("payment event loop disabled")
return
if not PAYMENT_CHECKOUT.enabled:
logger.info("payment event loop skipped: payment service disabled")
return
interval_sec = _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_INTERVAL_SEC", 20, 5)
lookback_blocks = _env_int(
"POLYWEATHER_PAYMENT_EVENT_LOOP_START_LOOKBACK_BLOCKS", 5000, 500
)
step_blocks = _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_STEP_BLOCKS", 2000, 100)
step_blocks = min(step_blocks, 49999)
max_events = _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_MAX_EVENTS_PER_CYCLE", 200, 10)
state_path = _state_file()
receiver_contracts = sorted(
{
_normalize_address(token.receiver_contract)
for token in PAYMENT_CHECKOUT.supported_tokens.values()
if _normalize_address(token.receiver_contract)
}
)
if not receiver_contracts:
logger.warning("payment event loop skipped: no receiver contract configured")
return
topic0 = str(PAYMENT_CHECKOUT._event_topic or "").strip().lower() # noqa: SLF001
if topic0 and not topic0.startswith("0x"):
topic0 = f"0x{topic0}"
if not topic0:
topic0 = (
"0x"
+ Web3.keccak(
text="OrderPaid(bytes32,address,uint256,address,uint256)"
).hex().lower().replace("0x", "")
)
logger.info(
"payment event loop started interval={}s lookback={} step={} max_events={} "
"contracts={} chain_id={}",
interval_sec,
lookback_blocks,
step_blocks,
max_events,
len(receiver_contracts),
PAYMENT_CHECKOUT.chain_id,
)
while True:
cycle_started = time.time()
try:
w3 = PAYMENT_CHECKOUT._get_web3() # noqa: SLF001
if not w3.is_connected():
logger.warning("payment event loop skipped: rpc not connected")
time.sleep(interval_sec)
continue
if int(w3.eth.chain_id) != int(PAYMENT_CHECKOUT.chain_id):
logger.warning(
"payment event loop skipped: chain mismatch rpc={} expected={}",
int(w3.eth.chain_id),
int(PAYMENT_CHECKOUT.chain_id),
)
time.sleep(interval_sec)
continue
latest_block = int(w3.eth.block_number)
safe_latest = latest_block - max(0, int(PAYMENT_CHECKOUT.confirmations) - 1)
if safe_latest <= 0:
time.sleep(interval_sec)
continue
state = _load_state(state_path)
last_scanned = int(state.get("last_scanned_block") or 0)
start_block = (
last_scanned + 1
if last_scanned > 0
else max(0, safe_latest - lookback_blocks + 1)
)
if start_block > safe_latest:
time.sleep(interval_sec)
continue
scanned_blocks = 0
scanned_events = 0
matched_intents = 0
submitted = 0
confirmed = 0
already = 0
pending = 0
failed = 0
ignored = 0
seen_events: set[str] = set()
cursor = start_block
while cursor <= safe_latest and scanned_events < max_events:
to_block = min(cursor + step_blocks - 1, safe_latest)
params: Dict[str, Any] = {
"fromBlock": cursor,
"toBlock": to_block,
"topics": [topic0],
}
params["address"] = (
receiver_contracts
if len(receiver_contracts) > 1
else receiver_contracts[0]
)
logs = w3.eth.get_logs(params)
scanned_blocks += max(0, to_block - cursor + 1)
state["last_scanned_block"] = to_block
state["updated_at"] = _now_iso()
_save_state(state_path, state)
if logs:
for log_item in logs:
event_row = _decode_order_paid_log(log_item)
if not event_row:
continue
event_key = _event_key(event_row)
if event_key in seen_events:
continue
seen_events.add(event_key)
scanned_events += 1
intents = PAYMENT_CHECKOUT.list_open_intents_by_order_id(
event_row["order_id_hex"],
limit=10,
)
if not intents:
ignored += 1
if scanned_events >= max_events:
break
continue
candidates = _select_matching_intents(intents, event_row)
for row in candidates:
status = str(row.get("status") or "").strip().lower()
if status == "confirmed":
already += 1
continue
user_id = str(row.get("user_id") or "").strip()
intent_id = str(row.get("intent_id") or "").strip()
if not user_id or not intent_id:
continue
matched_intents += 1
if status == "created" or not str(row.get("tx_hash") or "").strip():
try:
PAYMENT_CHECKOUT.submit_intent_tx(
user_id=user_id,
intent_id=intent_id,
tx_hash=event_row["tx_hash"],
from_address=event_row["payer"],
)
submitted += 1
except PaymentCheckoutError as exc:
if not _is_nonfatal_submit_error(exc):
failed += 1
logger.warning(
"payment event submit failed intent={} user={} tx={} status={} detail={}",
intent_id,
user_id,
event_row["tx_hash"],
exc.status_code,
exc.detail,
)
continue
try:
result = PAYMENT_CHECKOUT.confirm_intent_tx(
user_id=user_id,
intent_id=intent_id,
tx_hash=event_row["tx_hash"],
)
if bool(result.get("already_confirmed")):
already += 1
else:
confirmed += 1
logger.info(
"payment event-confirmed intent={} user={} tx={} block={}",
intent_id,
user_id,
event_row["tx_hash"],
int(event_row.get("block_number") or 0),
)
except PaymentCheckoutError as exc:
if _is_pending_confirm_error(exc):
pending += 1
continue
failed += 1
logger.warning(
"payment event confirm failed intent={} user={} tx={} status={} detail={}",
intent_id,
user_id,
event_row["tx_hash"],
exc.status_code,
exc.detail,
)
if scanned_events >= max_events:
break
cursor = to_block + 1
if scanned_blocks > 0:
logger.info(
"payment event cycle blocks={} events={} matched={} submitted={} "
"confirmed={} already={} pending={} failed={} ignored={}",
scanned_blocks,
scanned_events,
matched_intents,
submitted,
confirmed,
already,
pending,
failed,
ignored,
)
except Exception as exc:
logger.warning(f"payment event cycle failed: {exc}")
elapsed = time.time() - cycle_started
time.sleep(max(0.0, interval_sec - elapsed))
def start_payment_event_loop() -> threading.Thread:
thread = threading.Thread(
target=_runner,
daemon=True,
name="payment-event-loop",
)
thread.start()
return thread
src/utils/telegram_chat_ids.py
+20
View File
@@ -40,6 +40,26 @@ def get_telegram_chat_ids_from_env() -> List[str]:
)
def get_polymarket_wallet_activity_chat_ids_from_env() -> List[str]:
"""
Preferred envs for wallet activity push:
- POLYMARKET_WALLET_ACTIVITY_CHAT_IDS (comma-separated)
- POLYMARKET_WALLET_ACTIVITY_CHAT_ID (single)
Falls back to global TELEGRAM_CHAT_IDS / TELEGRAM_CHAT_ID for compatibility.
"""
dedicated = parse_telegram_chat_ids(
os.getenv("POLYMARKET_WALLET_ACTIVITY_CHAT_IDS"),
os.getenv("POLYMARKET_WALLET_ACTIVITY_CHAT_ID"),
)
if dedicated:
return dedicated
return parse_telegram_chat_ids(
os.getenv("TELEGRAM_CHAT_IDS"),
os.getenv("TELEGRAM_CHAT_ID"),
)
def get_primary_telegram_chat_id_from_env() -> str:
chat_ids = get_telegram_chat_ids_from_env()
return chat_ids[0] if chat_ids else ""