Монитор цен конкурентов

#!/usr/bin/env python3
"""Competitor price monitor: your products in rival shops, by item id.

Find an id once:   SATELIT_KEY=sat_live_... python3 price_monitor.py search "lapte sana"
Watch the list:    SATELIT_KEY=sat_live_... python3 price_monitor.py

Standard library only. Put it in cron: 0 7 * * * python3 /path/price_monitor.py
"""
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

sys.stdout.reconfigure(encoding="utf-8")  # names hold ă, ș, ț whatever the console's code page

BASE = os.environ.get("SATELIT_BASE", "https://api.satelit.md/v1")
KEY = os.environ["SATELIT_KEY"]
ALERT_URL = os.environ.get("ALERT_URL", "")  # a Slack-style incoming webhook; empty prints the alerts
MY_SHOP = os.environ.get("MY_SHOP", "")  # your shop's key in /v1/prices/stores, if you are listed
STATE = Path(os.environ.get("STATE_FILE", "price-state.json"))

# item id -> your own price, in lei
WATCH = {
    18342: 16.90,
}
credits_used = 0


def api(path, **params):
    global credits_used
    query = urllib.parse.urlencode({"lang": "ro", **params})
    req = urllib.request.Request(f"{BASE}{path}?{query}", headers={"Authorization": f"Bearer {KEY}"})
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=20) as res:
                body = json.load(res)
                credits_used += body["meta"].get("credits", 0)
                return body["data"]
        except urllib.error.HTTPError as err:
            retry = err.headers.get("Retry-After")
            if err.code in (429, 503) and retry and attempt < 2:
                time.sleep(int(retry))
                continue
            raise SystemExit(f"{path}: {err.code} {err.read().decode()}")


def alert(text):
    print(text)
    if ALERT_URL:
        req = urllib.request.Request(ALERT_URL, json.dumps({"text": text}).encode(), {"Content-Type": "application/json"})
        urllib.request.urlopen(req, timeout=10).close()


def search(query):
    for item in api("/prices/search", q=query):
        print(item["id"], item["name"], item["min_price"], f"{item['stores']} shops")


def trend(item_id, store):
    points = [p for p in api(f"/prices/items/{item_id}/history", days=30, store=store)["points"] if p["store"] == store]
    if len(points) < 2:
        return ""
    return f" (30 days: {points[0]['price']} to {points[-1]['price']})"


def main():
    state = json.loads(STATE.read_text()) if STATE.exists() else {}
    for item_id, my_price in WATCH.items():
        item = api(f"/prices/items/{item_id}")
        rivals = [o for o in item["offers"] if o["in_stock"] and o["store"]["key"] != MY_SHOP]
        seen = state.get(str(item_id), {})
        for offer in rivals:
            key, price = offer["store"]["key"], offer["price"]
            if key in seen and seen[key] != price:
                alert(f"{item['name']}: {offer['store']['name']} {seen[key]} -> {price} lei{trend(item_id, key)}")
            seen[key] = price
        state[str(item_id)] = seen
        if rivals:
            cheapest = min(rivals, key=lambda o: o["price"])
            if cheapest["price"] < my_price:
                alert(f"{item['name']}: {cheapest['store']['name']} {cheapest['price']} lei, you {my_price} lei. {cheapest['url']}")
    STATE.write_text(json.dumps(state))
    print(f"{len(WATCH)} items checked, {credits_used} credits")


if __name__ == "__main__":
    if len(sys.argv) > 2 and sys.argv[1] == "search":
        search(sys.argv[2])
    else:
        main()

#!/usr/bin/env node
// Receives price.changed from Satelit (Business and Enterprise), checks the signature, prints the change.
//   SATELIT_WEBHOOK_SECRET=... PORT=8080 node price_webhook.mjs
// Put it behind HTTPS and add the address in the dashboard, with your item ids.
import { createHmac, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";

const SECRET = process.env.SATELIT_WEBHOOK_SECRET;
if (!SECRET) throw new Error("Set SATELIT_WEBHOOK_SECRET");
const seen = new Set();

function verified(rawBody, header = "") {
  const parts = Object.fromEntries(header.split(",").map((p) => p.split("=")));
  const t = Number(parts.t);
  if (!t || Math.abs(Date.now() / 1000 - t) > 300) return false;
  const expected = Buffer.from(createHmac("sha256", SECRET).update(`${t}.${rawBody}`).digest("hex"));
  const given = Buffer.from(parts.v1 ?? "");
  return expected.length === given.length && timingSafeEqual(expected, given);
}

const server = createServer((req, res) => {
  const chunks = [];
  req.on("data", (c) => chunks.push(c));
  req.on("end", () => {
    const raw = Buffer.concat(chunks).toString("utf8");
    if (!verified(raw, req.headers["satelit-signature"])) {
      res.writeHead(401).end();
      return;
    }
    const event = JSON.parse(raw);
    res.writeHead(200).end();
    if (seen.has(event.id)) return; // a retry of what we already handled
    seen.add(event.id);
    if (event.type === "price.changed") {
      const { item, store, old, new: price, at } = event.data;
      console.log(`${at}: item ${item} at ${store}: ${old} -> ${price} lei`);
    }
  });
});
server.listen(Number(process.env.PORT ?? 8080), () => console.log(`listening on ${server.address().port}`));

#!/usr/bin/env python3
"""Fuel price monitor for a network: rivals station by station, against the ANRE cap.

SATELIT_KEY=sat_live_... MY_NETWORK=bemol RIVALS=lukoil,rompetrol python3 fuel_monitor.py
Standard library only. Put it in cron: 30 13 * * * python3 /path/fuel_monitor.py
"""
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

sys.stdout.reconfigure(encoding="utf-8")  # names hold ă, ș, ț whatever the console's code page

BASE = os.environ.get("SATELIT_BASE", "https://api.satelit.md/v1")
KEY = os.environ["SATELIT_KEY"]
MY_NETWORK = os.environ.get("MY_NETWORK", "bemol")
RIVALS = os.environ.get("RIVALS", "lukoil,rompetrol").split(",")
FUEL = os.environ.get("FUEL", "a95")  # a95, diesel or lpg
GAP = float(os.environ.get("GAP", "0.05"))  # alert when a rival is this many lei cheaper
ALERT_URL = os.environ.get("ALERT_URL", "")  # a Slack-style incoming webhook; empty prints the alerts
STATE = Path(os.environ.get("STATE_FILE", "fuel-state.json"))
credits_used = 0


def api(path, **params):
    global credits_used
    query = urllib.parse.urlencode({"lang": "ro", **params})
    req = urllib.request.Request(f"{BASE}{path}?{query}", headers={"Authorization": f"Bearer {KEY}"})
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=20) as res:
                body = json.load(res)
                credits_used += body["meta"].get("credits", 0)
                return body["data"], body["meta"]
        except urllib.error.HTTPError as err:
            retry = err.headers.get("Retry-After")
            if err.code in (429, 503) and retry and attempt < 2:
                time.sleep(int(retry))
                continue
            raise SystemExit(f"{path}: {err.code} {err.read().decode()}")


def alert(text):
    print(text)
    if ALERT_URL:
        req = urllib.request.Request(ALERT_URL, json.dumps({"text": text}).encode(), {"Content-Type": "application/json"})
        urllib.request.urlopen(req, timeout=10).close()


def stations(network):
    rows, page = [], 1
    while True:
        data, meta = api("/fuel/stations", network=network, page=page)
        rows += data
        if page >= meta["page"]["pages"]:
            return rows
        page += 1


def by_raion(rows):
    out = {}
    for s in rows:
        price = (s.get("prices") or {}).get(FUEL)
        if price is not None:
            out.setdefault(s["raion"], []).append((price, s))
    return out


def main():
    state = json.loads(STATE.read_text()) if STATE.exists() else {}
    board, _ = api("/fuel")
    cap = board["cap"].get(FUEL)

    if cap is not None and state.get("cap") not in (None, cap):
        alert(f"ANRE cap {FUEL}: {state['cap']} -> {cap} lei/l")
    state["cap"] = cap

    mine = by_raion(stations(MY_NETWORK))
    rivals = {}
    for rival in RIVALS:
        for raion, rows in by_raion(stations(rival)).items():
            rivals.setdefault(raion, []).extend((price, rival, s) for price, s in rows)

    for raion, own in sorted(mine.items()):
        for price, station in own:
            if cap is not None and price > cap:
                alert(f"Over the cap: {station['name']}, {raion}: {price} lei/l, cap {cap}")
            cheaper = [r for r in rivals.get(raion, []) if r[0] <= price - GAP]
            if cheaper:
                best = min(cheaper, key=lambda r: r[0])
                alert(f"{raion}: {best[1]} {best[2]['name']} {best[0]} lei/l, your {station['name']} {price} lei/l")

    for rival in RIVALS:
        data, _ = api("/fuel/history", fuel=FUEL, network=rival, days=30)
        points = data["points"]
        if len(points) > 1:
            print(f"{rival}: {points[0]['value']} to {points[-1]['value']} lei/l in {len(points)} days")

    STATE.write_text(json.dumps(state))
    print(f"{len(mine)} raions compared, {credits_used} credits")


if __name__ == "__main__":
    main()