feat: enhance CSV handling with backup snapshots and last updated date parsing
- Added support for parsing multiple date formats for the "lastUpdatedOn" field from Google Play. - Implemented a backup mechanism in CsvSink to create snapshots of the CSV file every N rows, with a configurable number of backups to keep. - Updated the CsvSink class to handle file flushing and backup creation safely. - Modified the run function to allow configuration of backup settings via command-line arguments.
This commit is contained in:
+84
-6
@@ -36,6 +36,8 @@ import argparse
|
||||
import csv
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
@@ -74,6 +76,33 @@ def epoch_to_date(ts):
|
||||
return None
|
||||
|
||||
|
||||
# Google Play's displayed "lastUpdatedOn" string can appear in several locale
|
||||
# formats. We parse it to ISO (YYYY-MM-DD) because it matches what the store
|
||||
# shows; the raw epoch is only a fallback (it can be off by a day in UTC).
|
||||
_PLAY_DATE_FORMATS = (
|
||||
"%b %d, %Y", # Jun 16, 2026
|
||||
"%B %d, %Y", # June 16, 2026
|
||||
"%d %b %Y", # 16 Jun 2026
|
||||
"%d %B %Y", # 16 June 2026
|
||||
"%Y-%m-%d", # 2026-06-16
|
||||
"%d-%m-%Y", # 16-06-2026
|
||||
"%m/%d/%Y", # 06/16/2026
|
||||
"%d/%m/%Y", # 16/06/2026
|
||||
)
|
||||
|
||||
|
||||
def play_last_updated(text, epoch):
|
||||
"""Return the last-updated date as ISO YYYY-MM-DD."""
|
||||
if text:
|
||||
t = str(text).strip()
|
||||
for fmt in _PLAY_DATE_FORMATS:
|
||||
try:
|
||||
return datetime.strptime(t, fmt).strftime("%Y-%m-%d")
|
||||
except ValueError:
|
||||
continue
|
||||
return epoch_to_date(epoch) # fallback when the string can't be parsed
|
||||
|
||||
|
||||
# --- Google Play ------------------------------------------------------------
|
||||
def play_row(app_id: str, lang: str, country: str) -> dict:
|
||||
d = gp_app(app_id, lang=lang, country=country)
|
||||
@@ -90,7 +119,7 @@ def play_row(app_id: str, lang: str, country: str) -> dict:
|
||||
"avg_rating": d.get("score"),
|
||||
"total_ratings": d.get("ratings"),
|
||||
"text_review_count": d.get("reviews"),
|
||||
"last_updated": d.get("lastUpdatedOn") or epoch_to_date(d.get("updated")),
|
||||
"last_updated": play_last_updated(d.get("lastUpdatedOn"), d.get("updated")),
|
||||
"version": d.get("version"),
|
||||
"url": d.get("url"),
|
||||
}
|
||||
@@ -134,9 +163,16 @@ def appstore_row(app_id: int, country: str) -> dict | None:
|
||||
class CsvSink:
|
||||
"""Append rows one at a time, flushing each so progress survives a crash."""
|
||||
|
||||
def __init__(self, path: Path, fields: list):
|
||||
def __init__(self, path: Path, fields: list,
|
||||
backup_every: int = 1000, keep: int = 5):
|
||||
self.path = path
|
||||
self.fields = fields
|
||||
self.lock = threading.Lock()
|
||||
self.backup_every = max(0, backup_every)
|
||||
self.keep = max(1, keep)
|
||||
self.backup_dir = path.parent / "backups"
|
||||
self.n = 0 # rows written this session
|
||||
self._bak_idx = -1 # ring-buffer index for rotating snapshots
|
||||
write_header = not path.exists() or path.stat().st_size == 0
|
||||
self.fh = path.open("a", newline="", encoding="utf-8-sig")
|
||||
self.writer = csv.DictWriter(self.fh, fieldnames=fields, extrasaction="ignore")
|
||||
@@ -147,10 +183,35 @@ class CsvSink:
|
||||
def write(self, row: dict) -> None:
|
||||
with self.lock:
|
||||
self.writer.writerow({k: row.get(k) for k in self.fields})
|
||||
self.fh.flush() # row is on disk immediately
|
||||
self.n += 1
|
||||
if self.backup_every and self.n % self.backup_every == 0:
|
||||
self._snapshot()
|
||||
|
||||
def _snapshot(self) -> None:
|
||||
"""Force to disk, then copy the live CSV into backups/ (ring of `keep`)."""
|
||||
try:
|
||||
self.fh.flush()
|
||||
os.fsync(self.fh.fileno()) # guarantee physical write
|
||||
self.backup_dir.mkdir(parents=True, exist_ok=True)
|
||||
self._bak_idx = (self._bak_idx + 1) % self.keep
|
||||
dest = self.backup_dir / f"{self.path.stem}.bak{self._bak_idx + 1}.csv"
|
||||
shutil.copyfile(self.path, dest)
|
||||
log.info("backup snapshot -> %s", dest)
|
||||
except Exception as exc: # never let a backup break the run
|
||||
log.warning("backup snapshot failed (continuing): %s", exc)
|
||||
|
||||
def close(self) -> None:
|
||||
self.fh.close()
|
||||
with self.lock:
|
||||
if self.backup_every:
|
||||
self._snapshot() # final snapshot before closing
|
||||
else:
|
||||
try:
|
||||
self.fh.flush()
|
||||
os.fsync(self.fh.fileno())
|
||||
except Exception:
|
||||
pass
|
||||
self.fh.close()
|
||||
|
||||
|
||||
# --- orchestration ----------------------------------------------------------
|
||||
@@ -220,7 +281,8 @@ def fetch_one(job, lang: str, delay: float, throttle: Throttle):
|
||||
|
||||
|
||||
def run(config_path: Path, out_dir: Path, countries_override: list | None,
|
||||
workers: int, shard: int, shards: int, resume: bool) -> None:
|
||||
workers: int, shard: int, shards: int, resume: bool,
|
||||
backup_every: int, backups_keep: int) -> None:
|
||||
if not config_path.exists():
|
||||
log.error("Config %s not found — run discover.py first to create it.",
|
||||
config_path)
|
||||
@@ -239,6 +301,17 @@ def run(config_path: Path, out_dir: Path, countries_override: list | None,
|
||||
name = "app_data.csv" if shards == 1 else f"app_data_shard{shard}of{shards}.csv"
|
||||
out_path = out_dir / name
|
||||
|
||||
# If the live CSV is gone but snapshots exist, the user can recover instead
|
||||
# of re-scraping from zero.
|
||||
backup_dir = out_dir / "backups"
|
||||
if not out_path.exists() and backup_dir.exists():
|
||||
baks = sorted(backup_dir.glob(f"{out_path.stem}.bak*.csv"),
|
||||
key=lambda p: p.stat().st_mtime, reverse=True)
|
||||
if baks:
|
||||
log.warning("%s is missing but a backup exists: %s", out_path, baks[0])
|
||||
log.warning("To resume instead of restarting, copy that backup back to "
|
||||
"%s, then re-run.", out_path)
|
||||
|
||||
skipped = 0
|
||||
if resume:
|
||||
done = load_done(out_path)
|
||||
@@ -248,7 +321,7 @@ def run(config_path: Path, out_dir: Path, countries_override: list | None,
|
||||
|
||||
total = len(jobs)
|
||||
throttle = Throttle()
|
||||
sink = CsvSink(out_path, CSV_FIELDS)
|
||||
sink = CsvSink(out_path, CSV_FIELDS, backup_every=backup_every, keep=backups_keep)
|
||||
|
||||
log.info(
|
||||
"shard %d/%d | %d jobs (%d already done, skipped) | %d workers | %s",
|
||||
@@ -308,6 +381,10 @@ def main() -> None:
|
||||
help="total number of machines splitting the work")
|
||||
parser.add_argument("--no-resume", action="store_true",
|
||||
help="ignore existing CSV and scrape everything again")
|
||||
parser.add_argument("--backup-every", type=int, default=1000,
|
||||
help="snapshot the CSV to backups/ every N rows (0 = off)")
|
||||
parser.add_argument("--backups-keep", type=int, default=5,
|
||||
help="how many rolling snapshots to keep")
|
||||
args = parser.parse_args()
|
||||
|
||||
if not 0 <= args.shard < args.shards:
|
||||
@@ -318,7 +395,8 @@ def main() -> None:
|
||||
if args.countries else None
|
||||
)
|
||||
run(Path(args.config), Path(args.out), countries,
|
||||
args.workers, args.shard, args.shards, resume=not args.no_resume)
|
||||
args.workers, args.shard, args.shards, resume=not args.no_resume,
|
||||
backup_every=args.backup_every, backups_keep=args.backups_keep)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user