monster
09972ccde1
A no-Xcode AppKit app (swiftc + build.sh): - Plaintext editor: 5 themes, line numbers, soft wrap, font controls, find, live Markdown preview (split mode), selection-aware Markdown toolbar. - Freeform Board mode: zoomable/pannable canvas of page cards (portrait/landscape, US Letter/A4/Legal/Square/Wide/Note), clean boxes, images (paste/drop/panel), freehand pen/shapes, screen-grab eyedropper, nested right-click menu, rich status bar. The Editor follows the selected page; layout persists to a foo.txt.board sidecar; the .txt stays pure plaintext. - scp-on-save to a Tailscale Mac (key auth, in-flight/quit drain). - HTML scraping pipeline: custom selector engine + Discogs profile, ingests to Postgres (psql) into a marketplace_snapshots schema. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
429 lines
18 KiB
Swift
429 lines
18 KiB
Swift
import Foundation
|
|
import CryptoKit
|
|
|
|
enum PGSettings {
|
|
static let autoIngestKey = "pgAutoIngest"
|
|
static let uriKey = "pgURI"
|
|
static let defaultURI = "postgresql://localhost:5433/scraperrr"
|
|
|
|
static var autoIngest: Bool { UserDefaults.standard.bool(forKey: autoIngestKey) }
|
|
static var uri: String {
|
|
let v = (UserDefaults.standard.string(forKey: uriKey) ?? "")
|
|
.trimmingCharacters(in: .whitespacesAndNewlines)
|
|
return v.isEmpty ? defaultURI : v
|
|
}
|
|
}
|
|
|
|
struct PGError: Error {
|
|
let message: String
|
|
}
|
|
|
|
/// Talks to Postgres by piping SQL into `psql` — no client library to link.
|
|
/// All values are escaped as SQL literals; numeric ids are validated in Swift.
|
|
enum PostgresStore {
|
|
|
|
static var psqlPath: String? {
|
|
let candidates = [
|
|
"/opt/homebrew/opt/postgresql@17/bin/psql",
|
|
"/opt/homebrew/opt/postgresql@18/bin/psql",
|
|
"/opt/homebrew/opt/postgresql@16/bin/psql",
|
|
"/opt/homebrew/bin/psql",
|
|
"/usr/local/bin/psql",
|
|
"/usr/bin/psql",
|
|
]
|
|
return candidates.first { FileManager.default.isExecutableFile(atPath: $0) }
|
|
}
|
|
|
|
struct PageIngest {
|
|
let url: String?
|
|
let sourceName: String
|
|
let contentHash: String
|
|
let elementCount: Int
|
|
let selectors: [(selector: String, count: Int, sample: String)]
|
|
let profileName: String?
|
|
let header: [String]
|
|
let rows: [[String]]
|
|
}
|
|
|
|
struct IngestReport {
|
|
let pageId: Int64
|
|
let selectorCount: Int
|
|
let rowCount: Int
|
|
let snapshotCount: Int
|
|
let skippedRows: Int
|
|
let skippedDuplicate: Bool
|
|
}
|
|
|
|
static func sha256(_ s: String) -> String {
|
|
SHA256.hash(data: Data(s.utf8)).map { String(format: "%02x", $0) }.joined()
|
|
}
|
|
|
|
// MARK: - Public entry points (call from main; completions on main)
|
|
|
|
static func test(uri: String, completion: @escaping (Result<String, PGError>) -> Void) {
|
|
let sql = schemaSQL + "\nSELECT current_database() || ' · ' || split_part(version(), ' on ', 1);"
|
|
runSQL(uri: uri, sql: sql) { result in
|
|
switch result {
|
|
case .success(let out):
|
|
let last = out.split(separator: "\n").last.map(String.init) ?? "ok"
|
|
completion(.success("Connected, tables ready — \(last)"))
|
|
case .failure(let error):
|
|
completion(.failure(error))
|
|
}
|
|
}
|
|
}
|
|
|
|
static func ingest(_ page: PageIngest, force: Bool,
|
|
completion: @escaping (Result<IngestReport, PGError>) -> Void) {
|
|
let uri = PGSettings.uri
|
|
|
|
func runMainIngest() {
|
|
let built = buildIngestSQL(page)
|
|
runSQL(uri: uri, sql: built.sql) { result in
|
|
switch result {
|
|
case .success(let out):
|
|
guard let idLine = out.split(separator: "\n").last,
|
|
let pageId = Int64(idLine.trimmingCharacters(in: .whitespaces)) else {
|
|
completion(.failure(PGError(message: "Couldn't read the new page id from psql output: \(out.suffix(120))")))
|
|
return
|
|
}
|
|
completion(.success(IngestReport(pageId: pageId,
|
|
selectorCount: page.selectors.count,
|
|
rowCount: page.rows.count,
|
|
snapshotCount: built.snapshotCount,
|
|
skippedRows: built.skipped,
|
|
skippedDuplicate: false)))
|
|
case .failure(let error):
|
|
completion(.failure(error))
|
|
}
|
|
}
|
|
}
|
|
|
|
if force || page.url == nil {
|
|
runMainIngest()
|
|
return
|
|
}
|
|
// Auto mode: skip if the latest capture of this URL has identical content.
|
|
let checkSQL = schemaSQL + """
|
|
|
|
SELECT content_hash FROM tp_pages WHERE url = \(lit(page.url))
|
|
ORDER BY captured_at DESC LIMIT 1;
|
|
"""
|
|
runSQL(uri: uri, sql: checkSQL) { result in
|
|
switch result {
|
|
case .success(let out):
|
|
let lastHash = out.split(separator: "\n").last.map(String.init) ?? ""
|
|
if lastHash == page.contentHash {
|
|
completion(.success(IngestReport(pageId: -1, selectorCount: 0, rowCount: 0,
|
|
snapshotCount: 0, skippedRows: 0, skippedDuplicate: true)))
|
|
} else {
|
|
runMainIngest()
|
|
}
|
|
case .failure(let error):
|
|
completion(.failure(error))
|
|
}
|
|
}
|
|
}
|
|
|
|
// MARK: - SQL construction
|
|
|
|
private static let schemaSQL = """
|
|
CREATE TABLE IF NOT EXISTS tp_pages (
|
|
id BIGSERIAL PRIMARY KEY,
|
|
url TEXT,
|
|
source_name TEXT,
|
|
content_hash TEXT,
|
|
element_count INT,
|
|
selector_count INT,
|
|
captured_at TIMESTAMPTZ DEFAULT now()
|
|
);
|
|
CREATE TABLE IF NOT EXISTS tp_selectors (
|
|
id BIGSERIAL PRIMARY KEY,
|
|
page_id BIGINT REFERENCES tp_pages(id) ON DELETE CASCADE,
|
|
selector TEXT NOT NULL,
|
|
match_count INT NOT NULL,
|
|
sample TEXT
|
|
);
|
|
CREATE TABLE IF NOT EXISTS tp_rows (
|
|
id BIGSERIAL PRIMARY KEY,
|
|
page_id BIGINT REFERENCES tp_pages(id) ON DELETE CASCADE,
|
|
profile TEXT,
|
|
row_num INT,
|
|
fields JSONB
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_tp_pages_url ON tp_pages(url);
|
|
CREATE INDEX IF NOT EXISTS idx_tp_rows_page ON tp_rows(page_id);
|
|
CREATE TABLE IF NOT EXISTS discogs_sellers (
|
|
seller_id BIGINT PRIMARY KEY,
|
|
username VARCHAR(100) NOT NULL,
|
|
last_scraped_at TIMESTAMPTZ DEFAULT now()
|
|
);
|
|
-- A username can be recycled across seller_ids over time; the seller_id is
|
|
-- the real key. Drop the old UNIQUE(username) so a reused name can't abort
|
|
-- a whole page ingest. (IF NOT EXISTS skips the table on existing installs,
|
|
-- so the constraint must be dropped explicitly.)
|
|
ALTER TABLE discogs_sellers DROP CONSTRAINT IF EXISTS discogs_sellers_username_key;
|
|
CREATE TABLE IF NOT EXISTS marketplace_snapshots (
|
|
item_id BIGINT PRIMARY KEY,
|
|
release_id BIGINT NOT NULL,
|
|
seller_id BIGINT REFERENCES discogs_sellers(seller_id),
|
|
title TEXT,
|
|
listing_url TEXT,
|
|
media_condition VARCHAR(120),
|
|
sleeve_condition VARCHAR(120),
|
|
price_currency VARCHAR(10),
|
|
price_value NUMERIC(12,2),
|
|
shipping_value NUMERIC(12,2),
|
|
first_seen TIMESTAMPTZ DEFAULT now(),
|
|
last_seen TIMESTAMPTZ DEFAULT now(),
|
|
is_active BOOLEAN DEFAULT TRUE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_market_release_id ON marketplace_snapshots(release_id);
|
|
CREATE INDEX IF NOT EXISTS idx_market_seller_active ON marketplace_snapshots(seller_id) WHERE is_active = TRUE;
|
|
"""
|
|
|
|
private static func buildIngestSQL(_ page: PageIngest) -> (sql: String, snapshotCount: Int, skipped: Int) {
|
|
var sql = schemaSQL + "\nBEGIN;\n"
|
|
sql += """
|
|
INSERT INTO tp_pages (url, source_name, content_hash, element_count, selector_count)
|
|
VALUES (\(lit(page.url)), \(lit(page.sourceName)), \(lit(page.contentHash)), \(page.elementCount), \(page.selectors.count))
|
|
RETURNING id AS page_id \\gset
|
|
|
|
"""
|
|
|
|
for chunk in page.selectors.chunked(into: 500) {
|
|
let values = chunk.map {
|
|
"(:page_id, \(lit($0.selector)), \($0.count), \(lit(String($0.sample.prefix(300)))))"
|
|
}.joined(separator: ",\n")
|
|
sql += "INSERT INTO tp_selectors (page_id, selector, match_count, sample) VALUES\n\(values);\n"
|
|
}
|
|
|
|
if !page.rows.isEmpty, let profile = page.profileName {
|
|
for (chunkIndex, chunk) in page.rows.chunked(into: 200).enumerated() {
|
|
let values = chunk.enumerated().map { offset, row -> String in
|
|
let rowNum = chunkIndex * 200 + offset + 1
|
|
var dict: [String: String] = [:]
|
|
for (i, name) in page.header.enumerated() where i < row.count {
|
|
// Condition fields capture the grade plus its tooltip
|
|
// description; keep the JSONB mirror as clean as the
|
|
// snapshot column (grade through the closing paren).
|
|
if name.hasSuffix("_condition"), let close = row[i].range(of: ")") {
|
|
dict[name] = String(row[i][..<close.upperBound])
|
|
} else {
|
|
dict[name] = row[i]
|
|
}
|
|
}
|
|
let json = (try? JSONSerialization.data(withJSONObject: dict))
|
|
.flatMap { String(data: $0, encoding: .utf8) } ?? "{}"
|
|
return "(:page_id, \(lit(profile)), \(rowNum), \(lit(json))::jsonb)"
|
|
}.joined(separator: ",\n")
|
|
sql += "INSERT INTO tp_rows (page_id, profile, row_num, fields) VALUES\n\(values);\n"
|
|
}
|
|
}
|
|
|
|
var snapshotCount = 0
|
|
var skipped = 0
|
|
if page.profileName == ScrapeProfile.discogsMarketplace.name {
|
|
let upserts = buildDiscogsUpserts(header: page.header, rows: page.rows)
|
|
sql += upserts.sql
|
|
snapshotCount = upserts.snapshotCount
|
|
skipped = upserts.skipped
|
|
}
|
|
|
|
sql += "COMMIT;\nSELECT :page_id;\n"
|
|
return (sql, snapshotCount, skipped)
|
|
}
|
|
|
|
/// Maps profile rows onto the scraperrr snapshot schema:
|
|
/// sellers upserted by data-seller-id, listings upserted by data-item-id
|
|
/// with last_seen/price refreshed on re-capture.
|
|
private static func buildDiscogsUpserts(header: [String], rows: [[String]]) -> (sql: String, snapshotCount: Int, skipped: Int) {
|
|
func col(_ name: String) -> Int? { header.firstIndex(of: name) }
|
|
guard let itemIdx = col("item_id"), let relIdx = col("release_id") else { return ("", 0, 0) }
|
|
let sellerIdIdx = col("seller_id"), sellerNameIdx = col("seller_username")
|
|
let sellerNameAltIdx = col("seller_username_alt")
|
|
let titleIdx = col("title"), urlIdx = col("listing_url")
|
|
let mediaIdx = col("media_condition"), sleeveIdx = col("sleeve_condition")
|
|
let curIdx = col("price_currency"), priceIdx = col("price_value")
|
|
let shipIdx = col("shipping")
|
|
|
|
func value(_ row: [String], _ idx: Int?) -> String? {
|
|
guard let idx, idx < row.count else { return nil }
|
|
let v = row[idx].trimmingCharacters(in: .whitespaces)
|
|
return v.isEmpty ? nil : v
|
|
}
|
|
|
|
// "Near Mint (NM or M-) A nearly perfect record…" → keep through the ")".
|
|
func condition(_ s: String?) -> String? {
|
|
guard let s else { return nil }
|
|
if let close = s.range(of: ")") { return String(s[..<close.upperBound]) }
|
|
return String(s.prefix(60))
|
|
}
|
|
|
|
func numeric(_ s: String?) -> String? {
|
|
guard let s, let d = Double(s) else { return nil }
|
|
return String(format: "%.2f", d)
|
|
}
|
|
|
|
func firstDecimal(_ s: String?) -> String? {
|
|
guard let s, let r = s.range(of: #"\d+(\.\d+)?"#, options: .regularExpression) else { return nil }
|
|
return numeric(String(s[r]))
|
|
}
|
|
|
|
var sellers: [Int64: String] = [:]
|
|
var snapshots: [Int64: String] = [:] // item_id → VALUES tuple (dedup within batch)
|
|
var skipped = 0
|
|
|
|
for row in rows {
|
|
guard let itemId = value(row, itemIdx).flatMap({ Int64($0) }),
|
|
let releaseId = value(row, relIdx).flatMap({ Int64($0) }) else {
|
|
skipped += 1
|
|
continue
|
|
}
|
|
// Only reference a seller_id the snapshot's FK can resolve: upsert
|
|
// the seller whenever its id is present (falling back to the
|
|
// co-located data-username, then a derived name), and NULL the
|
|
// snapshot's seller_id when no id was found.
|
|
var linkedSellerId: Int64?
|
|
if let sellerId = value(row, sellerIdIdx).flatMap({ Int64($0) }) {
|
|
let name = value(row, sellerNameIdx)
|
|
?? value(row, sellerNameAltIdx)
|
|
?? "seller_\(sellerId)"
|
|
sellers[sellerId] = name
|
|
linkedSellerId = sellerId
|
|
}
|
|
let tuple = """
|
|
(\(itemId), \(releaseId), \(linkedSellerId.map(String.init) ?? "NULL"), \
|
|
\(lit(value(row, titleIdx))), \(lit(value(row, urlIdx))), \
|
|
\(lit(condition(value(row, mediaIdx)))), \(lit(condition(value(row, sleeveIdx)))), \
|
|
\(lit(value(row, curIdx))), \(numeric(value(row, priceIdx)) ?? "NULL"), \
|
|
\(firstDecimal(value(row, shipIdx)) ?? "NULL"))
|
|
"""
|
|
snapshots[itemId] = tuple
|
|
}
|
|
guard !snapshots.isEmpty else { return ("", 0, skipped) }
|
|
|
|
var sql = ""
|
|
if !sellers.isEmpty {
|
|
let sellerValues = sellers
|
|
.sorted { $0.key < $1.key }
|
|
.map { "(\($0.key), \(lit(String($0.value.prefix(100)))))" }
|
|
.joined(separator: ",\n")
|
|
sql += """
|
|
INSERT INTO discogs_sellers (seller_id, username) VALUES
|
|
\(sellerValues)
|
|
ON CONFLICT (seller_id) DO UPDATE SET username = EXCLUDED.username, last_scraped_at = now();
|
|
|
|
"""
|
|
}
|
|
let snapshotValues = snapshots.sorted { $0.key < $1.key }.map { $0.value }.joined(separator: ",\n")
|
|
sql += """
|
|
INSERT INTO marketplace_snapshots
|
|
(item_id, release_id, seller_id, title, listing_url, media_condition,
|
|
sleeve_condition, price_currency, price_value, shipping_value) VALUES
|
|
\(snapshotValues)
|
|
ON CONFLICT (item_id) DO UPDATE SET
|
|
last_seen = now(),
|
|
price_value = EXCLUDED.price_value,
|
|
shipping_value = EXCLUDED.shipping_value,
|
|
media_condition = EXCLUDED.media_condition,
|
|
sleeve_condition = EXCLUDED.sleeve_condition,
|
|
is_active = TRUE;
|
|
|
|
"""
|
|
return (sql, snapshots.count, skipped)
|
|
}
|
|
|
|
private static func lit(_ s: String?) -> String {
|
|
guard let s else { return "NULL" }
|
|
let cleaned = s.replacingOccurrences(of: "\0", with: "")
|
|
return "'" + cleaned.replacingOccurrences(of: "'", with: "''") + "'"
|
|
}
|
|
|
|
// MARK: - psql runner
|
|
|
|
private static let workQueue = DispatchQueue(label: "textplus.postgres")
|
|
private static let timeoutSeconds: TimeInterval = 60
|
|
|
|
private static func runSQL(uri: String, sql: String,
|
|
completion: @escaping (Result<String, PGError>) -> Void) {
|
|
guard let psql = psqlPath else {
|
|
DispatchQueue.main.async {
|
|
completion(.failure(PGError(message: "psql not found — install with: brew install postgresql@17")))
|
|
}
|
|
return
|
|
}
|
|
workQueue.async {
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: psql)
|
|
process.arguments = ["-X", "-q", "-A", "-t", "-v", "ON_ERROR_STOP=1", uri]
|
|
|
|
let inPipe = Pipe(), outPipe = Pipe(), errPipe = Pipe()
|
|
process.standardInput = inPipe
|
|
process.standardOutput = outPipe
|
|
process.standardError = errPipe
|
|
|
|
do {
|
|
try process.run()
|
|
} catch {
|
|
DispatchQueue.main.async { completion(.failure(PGError(message: error.localizedDescription))) }
|
|
return
|
|
}
|
|
|
|
let watchdog = DispatchWorkItem { [weak process] in
|
|
guard let process, process.isRunning else { return }
|
|
process.terminate()
|
|
}
|
|
DispatchQueue.global().asyncAfter(deadline: .now() + timeoutSeconds, execute: watchdog)
|
|
|
|
// Drain stdout/stderr concurrently while we feed stdin, so a large
|
|
// script can't deadlock on a full pipe in either direction.
|
|
var outData = Data()
|
|
var errData = Data()
|
|
let group = DispatchGroup()
|
|
group.enter()
|
|
DispatchQueue.global(qos: .utility).async {
|
|
outData = outPipe.fileHandleForReading.readDataToEndOfFile()
|
|
group.leave()
|
|
}
|
|
group.enter()
|
|
DispatchQueue.global(qos: .utility).async {
|
|
errData = errPipe.fileHandleForReading.readDataToEndOfFile()
|
|
group.leave()
|
|
}
|
|
|
|
let writer = inPipe.fileHandleForWriting
|
|
do {
|
|
try writer.write(contentsOf: Data(sql.utf8))
|
|
try writer.close()
|
|
} catch {
|
|
// Broken pipe (psql died early) — fall through to read its stderr.
|
|
try? writer.close()
|
|
}
|
|
|
|
group.wait()
|
|
process.waitUntilExit()
|
|
watchdog.cancel()
|
|
|
|
let out = String(decoding: outData, as: UTF8.self).trimmingCharacters(in: .whitespacesAndNewlines)
|
|
let err = String(decoding: errData, as: UTF8.self).trimmingCharacters(in: .whitespacesAndNewlines)
|
|
let status = process.terminationStatus
|
|
|
|
DispatchQueue.main.async {
|
|
if status == 0 {
|
|
completion(.success(out))
|
|
} else {
|
|
let detail = err.isEmpty ? "psql exited with status \(status)" : err
|
|
completion(.failure(PGError(message: detail)))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private extension Array {
|
|
func chunked(into size: Int) -> [[Element]] {
|
|
stride(from: 0, to: count, by: size).map { Array(self[$0..<Swift.min($0 + size, count)]) }
|
|
}
|
|
}
|