import os
import re
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from nstdata_ai_crawl import Format, NstDataClient, ScrapeRequestDto
DB_PATH = "prices.sqlite3"
PRODUCTS = [
{
"product_id": "widget-standard",
"product_name": "Widget Standard",
"source_url": "https://shop.example.com/products/widget",
"currency": "USD",
}
]
def fetch_markdown(client: NstDataClient, url: str) -> str:
result = client.submit_scrape_task_sync(
ScrapeRequestDto(
url=url,
formats=[Format.MARKDOWN],
onlyMainContent=True,
timeout=60000,
)
)
body = result.data.get_markdown()
if not body or not body.strip():
raise ValueError("没有可用的产品页面内容")
return body
def parse_price(markdown: str, currency: str) -> Decimal:
# 用经过测试的解析器替换此特定目标标记。
pattern = rf"(?m)^价格:\s*{re.escape(currency)}\s+([0-9]+(?:\.[0-9]{{2}})?)\s*$"
matches = re.findall(pattern, markdown)
if len(matches) != 1:
raise ValueError(f"期望找到一个 {currency} 价格标记，实际发现了 {len(matches)} 个")
try:
amount = Decimal(matches[0])
except InvalidOperation as exc:
raise ValueError("无效的十进制价格") from exc
if amount <= 0:
raise ValueError("价格必须为正")
return amount
def init_db(db: sqlite3.Connection) -> None:
db.execute("""
CREATE TABLE IF NOT EXISTS observations (
id INTEGER PRIMARY KEY,
product_id TEXT NOT NULL,
observed_price TEXT,
currency TEXT NOT NULL,
observed_at TEXT NOT NULL,
source_url TEXT NOT NULL,
extraction_status TEXT NOT NULL
)
""")
db.commit()
def previous_price(db: sqlite3.Connection, product_id: str, currency: str):
row = db.execute(
"""SELECT observed_price FROM observations
WHERE product_id = ? AND currency = ? AND extraction_status = 'ok'
ORDER BY id DESC LIMIT 1""",
(product_id, currency),
).fetchone()
return Decimal(row[0]) if row else None
def save(db, product, observed_at, status, price=None):
db.execute(
"""INSERT INTO observations
(product_id, observed_price, currency, observed_at, source_url, extraction_status)
VALUES (?, ?, ?, ?, ?, ?)""",
(
product["product_id"],
str(price) if price is not None else None,
product["currency"],
observed_at,
product["source_url"],
status,
),
)
db.commit()
def main() -> None:
with sqlite3.connect(DB_PATH) as db:
init_db(db)
with NstDataClient(os.environ["NSTDATA_API_TOKEN"]) as client:
for product in PRODUCTS:
observed_at = datetime.now(timezone.utc).isoformat()
try:
markdown = fetch_markdown(client, product["source_url"])
price = parse_price(markdown, product["currency"])
baseline = previous_price(
db, product["product_id"], product["currency"]
)
save(db, product, observed_at, "ok", price)
if baseline is None:
print(f"基线记录：{product['product_id']} {price}")
continue
delta = price - baseline
pct = abs(delta / baseline) * 100 if baseline else Decimal("0")
if abs(delta) >= Decimal("5") and pct >= Decimal("3"):
print(
f"警报 {product['product_id']}：{baseline} -> {price} "
f"{product['currency']} ({pct:.2f}%); "
f"来源={product['source_url']} at={observed_at}"
)
except Exception as exc:
save(db, product, observed_at, "error")
print(f"错误 {product['product_id']}：{exc}")
if __name__ == "__main__":
main()
