-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmodels.py
More file actions
312 lines (254 loc) · 11.1 KB
/
Copy pathmodels.py
File metadata and controls
312 lines (254 loc) · 11.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
import contextlib
import os
import sqlite3
from core.paths import data_dir as _paths_data_dir
DB_PATH = str(_paths_data_dir() / "shop_py_bot.db")
@contextlib.contextmanager
def get_db_connection():
"""Open a WAL-enabled connection, yield it, commit/rollback, close.
PRAGMA sequence (ASYNC-04, exact order required):
journal_mode=WAL -- readers never block writers under concurrent polling
busy_timeout=5000 -- retry for up to 5s on write-lock contention
synchronous=NORMAL -- safe with WAL; better throughput than FULL
"""
conn = sqlite3.connect(DB_PATH, timeout=5)
try:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=5000")
conn.execute("PRAGMA synchronous=NORMAL")
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def initialize_db(delete=False):
if delete and os.path.exists(DB_PATH):
os.remove(DB_PATH)
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
with get_db_connection() as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS items (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
link TEXT NOT NULL UNIQUE,
auto_buy BOOLEAN NOT NULL,
quantity INTEGER NOT NULL,
purchased BOOLEAN NOT NULL DEFAULT 0
)
''')
# Idempotent column additions: read existing columns first, add only if absent.
existing = {
row[1]
for row in conn.execute("PRAGMA table_info(items)").fetchall()
}
if "last_seen_available" not in existing:
conn.execute(
"ALTER TABLE items ADD COLUMN last_seen_available INTEGER NOT NULL DEFAULT 0"
)
if "last_notified" not in existing:
conn.execute(
"ALTER TABLE items ADD COLUMN last_notified TEXT"
)
# Phase 16: price monitoring columns (idempotent -- PRICE-01, PRICE-05).
if "target_price" not in existing:
conn.execute("ALTER TABLE items ADD COLUMN target_price INTEGER")
if "price_drop_pct" not in existing:
conn.execute("ALTER TABLE items ADD COLUMN price_drop_pct REAL")
if "price_alert_armed" not in existing:
conn.execute(
"ALTER TABLE items ADD COLUMN price_alert_armed INTEGER NOT NULL DEFAULT 0"
)
if "price_last_notified" not in existing:
conn.execute("ALTER TABLE items ADD COLUMN price_last_notified TEXT")
# Phase 19: confirmation columns (BUY-03, BUY-04).
if "order_id" not in existing:
conn.execute("ALTER TABLE items ADD COLUMN order_id TEXT")
if "confirmed_at" not in existing:
conn.execute("ALTER TABLE items ADD COLUMN confirmed_at TEXT")
if "checkout_attempts" not in existing:
conn.execute(
"ALTER TABLE items ADD COLUMN checkout_attempts INTEGER NOT NULL DEFAULT 0"
)
# Phase 16: append-only price history table (PRICE-02).
conn.execute('''
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY,
item_link TEXT NOT NULL,
price_cents INTEGER NOT NULL,
currency TEXT NOT NULL DEFAULT 'USD',
scraped_at TEXT NOT NULL
)
''')
def get_items_sync():
"""Return all items as a list of (name, link, auto_buy, quantity, purchased)."""
with get_db_connection() as conn:
return conn.execute(
"SELECT name, link, auto_buy, quantity, purchased FROM items"
).fetchall()
def update_item_purchased_sync(link):
"""Set purchased=1 for the item with the given link."""
with get_db_connection() as conn:
conn.execute("UPDATE items SET purchased=1 WHERE link=?", (link,))
def increment_checkout_attempts_sync(link: str) -> None:
"""Increment checkout_attempts by exactly 1 for the given link (BUY-05/REL-08).
Uses a relative SQL increment so each call adds 1 without knowing the current value.
Never resets the counter; monotonic per-attempt accounting for double-buy detection.
"""
with get_db_connection() as conn:
conn.execute(
"UPDATE items SET checkout_attempts = checkout_attempts + 1 WHERE link=?",
(link,),
)
def get_item_order_state_sync(link: str) -> tuple[bool, str | None]:
"""Return (purchased as bool, order_id) for the cart-retry idempotency check (BUY-05).
Returns (False, None) when the item row is missing (safe no-op for retry guard).
"""
with get_db_connection() as conn:
row = conn.execute(
"SELECT purchased, order_id FROM items WHERE link=?",
(link,),
).fetchone()
if row is None:
return False, None
return bool(row[0]), row[1]
def update_item_confirmed_sync(link: str, order_id: str, confirmed_at: str) -> None:
"""Set purchased=1, order_id, and confirmed_at together (BUY-03/BUY-04).
The order_id column is the idempotency anchor for Phase 21 retry (BUY-05).
Does not touch checkout_attempts (increment is Phase 21 scope).
"""
with get_db_connection() as conn:
conn.execute(
"UPDATE items SET purchased=1, order_id=?, confirmed_at=? WHERE link=?",
(order_id, confirmed_at, link),
)
def add_items_sync(items):
"""Insert items that are not already present (unique by link)."""
with get_db_connection() as conn:
for item in items:
count = conn.execute(
"SELECT COUNT(*) FROM items WHERE link=?", (item[1],)
).fetchone()[0]
if count == 0:
conn.execute(
"INSERT INTO items (name, link, auto_buy, quantity, purchased)"
" VALUES (?, ?, ?, ?, ?)",
item,
)
def remove_item_sync(link):
"""Delete the item with the given link (parameterized DELETE)."""
with get_db_connection() as conn:
conn.execute("DELETE FROM items WHERE link=?", (link,))
def get_item_notification_state_sync(link: str) -> tuple[bool, str | None]:
"""Return (last_seen_available as bool, last_notified) for dedup checks (NOTIF-02)."""
with get_db_connection() as conn:
row = conn.execute(
"SELECT last_seen_available, last_notified FROM items WHERE link=?",
(link,),
).fetchone()
if row is None:
return False, None
return bool(row[0]), row[1]
def set_item_available_sync(link: str, notified_at: str) -> None:
"""Set last_seen_available=1 and last_notified=notified_at for rising-edge dedup (NOTIF-02)."""
with get_db_connection() as conn:
conn.execute(
"UPDATE items SET last_seen_available=1, last_notified=? WHERE link=?",
(notified_at, link),
)
def clear_item_available_sync(link: str) -> None:
"""Set last_seen_available=0 (item went out of stock); preserve last_notified (NOTIF-02)."""
with get_db_connection() as conn:
conn.execute(
"UPDATE items SET last_seen_available=0 WHERE link=?",
(link,),
)
# ---------------------------------------------------------------------------
# Phase 16: price history + price-alert dedup + price config (PRICE-01/02/05)
# ---------------------------------------------------------------------------
def append_price_history_sync(
link: str, price_cents: int, scraped_at: str, currency: str = "USD"
) -> None:
"""Insert one price observation into the append-only price_history table."""
with get_db_connection() as conn:
conn.execute(
"INSERT INTO price_history (item_link, price_cents, currency, scraped_at)"
" VALUES (?, ?, ?, ?)",
(link, price_cents, currency, scraped_at),
)
def get_price_history_sync(link: str, limit: int = 10) -> list[tuple]:
"""Return last N (price_cents, currency, scraped_at) rows, newest first."""
with get_db_connection() as conn:
return conn.execute(
"SELECT price_cents, currency, scraped_at FROM price_history"
" WHERE item_link=? ORDER BY scraped_at DESC LIMIT ?",
(link, limit),
).fetchall()
def get_last_price_sync(link: str) -> int | None:
"""Return the most recent price_cents for the item, or None if no history."""
with get_db_connection() as conn:
row = conn.execute(
"SELECT price_cents FROM price_history"
" WHERE item_link=? ORDER BY scraped_at DESC LIMIT 1",
(link,),
).fetchone()
return row[0] if row else None
def get_price_alert_state_sync(link: str) -> tuple[bool, str | None]:
"""Return (price_alert_armed as bool, price_last_notified) for price dedup.
Returns (False, None) when the item row is missing.
"""
with get_db_connection() as conn:
row = conn.execute(
"SELECT price_alert_armed, price_last_notified FROM items WHERE link=?",
(link,),
).fetchone()
if row is None:
return False, None
return bool(row[0]), row[1]
def set_price_alert_armed_sync(link: str, notified_at: str) -> None:
"""Set price_alert_armed=1 and price_last_notified for price-drop dedup.
MUST NOT touch last_seen_available or last_notified (Pitfall 1).
"""
with get_db_connection() as conn:
conn.execute(
"UPDATE items SET price_alert_armed=1, price_last_notified=? WHERE link=?",
(notified_at, link),
)
def clear_price_alert_armed_sync(link: str) -> None:
"""Reset price_alert_armed=0 (price recovered above threshold)."""
with get_db_connection() as conn:
conn.execute(
"UPDATE items SET price_alert_armed=0 WHERE link=?",
(link,),
)
def update_item_price_config_sync(
link: str, target_price: int | None, price_drop_pct: float | None
) -> None:
"""Write per-item price monitoring config to the items table (PRICE-01/05).
Called after add_items_sync to seed config from ItemConfig fields.
Idempotent: NULL overwrites NULL when config has no price targets.
"""
with get_db_connection() as conn:
conn.execute(
"UPDATE items SET target_price=?, price_drop_pct=? WHERE link=?",
(target_price, price_drop_pct, link),
)
def get_item_price_config_sync(link: str) -> tuple[int | None, float | None] | None:
"""Return (target_price, price_drop_pct) for the item, or None if row missing."""
with get_db_connection() as conn:
row = conn.execute(
"SELECT target_price, price_drop_pct FROM items WHERE link=?",
(link,),
).fetchone()
if row is None:
return None
return row[0], row[1]
# Legacy names kept for backward compatibility (existing tests and imports).
# Single source of truth: body lives in the _sync functions above.
def get_items():
return get_items_sync()
def update_item_purchased(link):
return update_item_purchased_sync(link)
def add_items(items):
return add_items_sync(items)