Jason Thistlethwaite
579 lines
19 KiB
Python
Executable File
579 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""External worker for Redmine event outbox rows.
|
|
|
|
This worker runs outside Redmine. It claims pending `event_outbox_events` rows
|
|
from the Redmine database over SSH/MySQL, enriches them with read-only joins,
|
|
writes deterministic JSONL records, and marks rows processed only after the
|
|
write succeeds.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
|
|
DEFAULT_SSH_HOST = "reddev@192.168.50.170"
|
|
DEFAULT_SSH_KEY = Path("/tmp/reddev")
|
|
DEFAULT_REMOTE_REDMINE = "/usr/share/redmine"
|
|
DEFAULT_OUTPUT = Path("/tmp/redmine-outbox/derived_documents.jsonl")
|
|
|
|
|
|
class OutboxWorkerError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class RemoteRedmine:
|
|
def __init__(self, ssh_host, ssh_key, remote_redmine, local=False):
|
|
self.ssh_host = ssh_host
|
|
self.ssh_key = ssh_key
|
|
self.remote_redmine = remote_redmine
|
|
self.local = local
|
|
|
|
def mysql_json_lines(self, sql):
|
|
stdout = self.mysql(sql)
|
|
rows = []
|
|
for line in stdout.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
rows.append(json.loads(bytes.fromhex(line.strip()).decode("utf-8")))
|
|
except json.JSONDecodeError as exc:
|
|
raise OutboxWorkerError(f"Remote query returned non-JSON row: {line[:200]}") from exc
|
|
except ValueError as exc:
|
|
raise OutboxWorkerError(f"Remote query returned non-hex row: {line[:200]}") from exc
|
|
return rows
|
|
|
|
def mysql(self, sql):
|
|
command = self._mysql_runner_command()
|
|
shell = True
|
|
if not self.local:
|
|
command = [
|
|
"ssh",
|
|
"-i",
|
|
str(self.ssh_key),
|
|
"-o",
|
|
"IdentitiesOnly=yes",
|
|
self.ssh_host,
|
|
self._mysql_runner_command(),
|
|
]
|
|
shell = False
|
|
try:
|
|
result = subprocess.run(
|
|
command,
|
|
input=sql,
|
|
universal_newlines=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
shell=shell,
|
|
)
|
|
except OSError as exc:
|
|
raise OutboxWorkerError(f"Could not run ssh: {exc}") from exc
|
|
|
|
if result.returncode != 0:
|
|
raise OutboxWorkerError(result.stderr.strip() or "Remote MySQL command failed.")
|
|
return result.stdout
|
|
|
|
def _mysql_runner_command(self):
|
|
ruby = (
|
|
"require 'yaml'; "
|
|
"c = YAML.load_file('config/database.yml')['production']; "
|
|
"ENV['MYSQL_PWD'] = c['password'].to_s; "
|
|
"args = ['--batch', '--raw', '--quick', '--skip-column-names', "
|
|
"'--default-character-set=utf8', '-h', c['host'].to_s, "
|
|
"'-P', (c['port'] || 3306).to_s, '-u', c['username'].to_s, c['database'].to_s]; "
|
|
"exec('mysql', *args)"
|
|
)
|
|
return f"cd {shell_quote(self.remote_redmine)} && ruby -e {shell_quote(ruby)}"
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Process Redmine event outbox rows into enriched JSONL documents.")
|
|
parser.add_argument("--ssh-host", default=os.getenv("REDMINE_SSH_HOST", DEFAULT_SSH_HOST))
|
|
parser.add_argument("--ssh-key", type=Path, default=Path(os.getenv("REDMINE_SSH_KEY", str(DEFAULT_SSH_KEY))))
|
|
parser.add_argument("--local", action="store_true", help="Read the Redmine database locally instead of over SSH.")
|
|
parser.add_argument("--remote-redmine", default=os.getenv("REDMINE_REMOTE_PATH", DEFAULT_REMOTE_REDMINE))
|
|
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
|
parser.add_argument("--batch-size", type=int, default=20)
|
|
parser.add_argument("--max-attempts", type=int, default=5)
|
|
parser.add_argument("--stale-lock-minutes", type=int, default=30)
|
|
parser.add_argument("--dry-run", action="store_true", help="Fetch and enrich pending rows without locking or marking them.")
|
|
parser.add_argument("--claim-only", action="store_true", help="Claim rows, print them, then release the claim without marking processed.")
|
|
parser.add_argument("--status", action="store_true", help="Report outbox row counts and exit.")
|
|
parser.add_argument(
|
|
"--purge-processed-days",
|
|
type=int,
|
|
help="Preview purging processed rows older than this many days, or delete with --apply-purge.",
|
|
)
|
|
parser.add_argument("--apply-purge", action="store_true", help="Actually delete rows selected by --purge-processed-days.")
|
|
args = parser.parse_args()
|
|
|
|
remote = RemoteRedmine(args.ssh_host, args.ssh_key, args.remote_redmine, local=args.local)
|
|
worker_id = make_worker_id()
|
|
|
|
try:
|
|
if args.apply_purge and args.purge_processed_days is None:
|
|
raise OutboxWorkerError("--apply-purge requires --purge-processed-days.")
|
|
if args.status:
|
|
print(json.dumps(outbox_status(remote, args.max_attempts, args.stale_lock_minutes), indent=2, sort_keys=True))
|
|
return 0
|
|
if args.purge_processed_days is not None:
|
|
purge_count = purge_processed(remote, args.purge_processed_days, apply=args.apply_purge)
|
|
action = "purged" if args.apply_purge else "would purge"
|
|
print(f"{action} {purge_count} processed outbox row(s)")
|
|
return 0
|
|
|
|
if args.dry_run:
|
|
events = pending_events(remote, args.batch_size, args.max_attempts, args.stale_lock_minutes)
|
|
else:
|
|
events = claim_events(remote, worker_id, args.batch_size, args.max_attempts, args.stale_lock_minutes)
|
|
|
|
if args.claim_only:
|
|
print(json.dumps(events, indent=2, sort_keys=True))
|
|
if not args.dry_run:
|
|
release_claims(remote, worker_id)
|
|
return 0
|
|
|
|
processed = 0
|
|
for event in events:
|
|
try:
|
|
documents = enrich_event(remote, event)
|
|
if args.dry_run:
|
|
for document in documents:
|
|
print(json.dumps(document, ensure_ascii=False, sort_keys=True))
|
|
else:
|
|
append_jsonl(args.output, documents)
|
|
mark_processed(remote, event["id"], worker_id)
|
|
processed += 1
|
|
except Exception as exc:
|
|
if args.dry_run:
|
|
raise
|
|
mark_failed(remote, event["id"], worker_id, exc)
|
|
print(f"event #{event.get('id')} failed: {exc}", file=sys.stderr)
|
|
|
|
action = "previewed" if args.dry_run else "processed"
|
|
print(f"{action} {processed} outbox event(s)")
|
|
if processed and not args.dry_run:
|
|
print(f"output: {args.output}")
|
|
return 0
|
|
except OutboxWorkerError as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def pending_events(remote, limit, max_attempts, stale_lock_minutes):
|
|
return remote.mysql_json_lines(
|
|
f"""
|
|
SELECT HEX(CAST(JSON_OBJECT(
|
|
'id', id,
|
|
'event_type', event_type,
|
|
'source_type', source_type,
|
|
'source_id', source_id,
|
|
'project_id', project_id,
|
|
'issue_id', issue_id,
|
|
'journal_id', journal_id,
|
|
'user_id', user_id,
|
|
'occurred_at', DATE_FORMAT(occurred_at, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'attempts', attempts,
|
|
'payload', payload
|
|
) AS CHAR)) AS document
|
|
FROM event_outbox_events
|
|
WHERE processed_at IS NULL
|
|
AND attempts < {sql_int(max_attempts)}
|
|
AND (locked_at IS NULL OR locked_at < UTC_TIMESTAMP() - INTERVAL {sql_int(stale_lock_minutes)} MINUTE)
|
|
ORDER BY id
|
|
LIMIT {sql_int(limit)};
|
|
"""
|
|
)
|
|
|
|
|
|
def outbox_status(remote, max_attempts, stale_lock_minutes):
|
|
rows = remote.mysql_json_lines(
|
|
f"""
|
|
SELECT HEX(CAST(JSON_OBJECT(
|
|
'total', COUNT(*),
|
|
'pending', SUM(IF(processed_at IS NULL AND attempts < {sql_int(max_attempts)}, 1, 0)),
|
|
'ready', SUM(IF(processed_at IS NULL
|
|
AND attempts < {sql_int(max_attempts)}
|
|
AND (locked_at IS NULL OR locked_at < UTC_TIMESTAMP() - INTERVAL {sql_int(stale_lock_minutes)} MINUTE), 1, 0)),
|
|
'locked', SUM(IF(processed_at IS NULL
|
|
AND locked_at IS NOT NULL
|
|
AND locked_at >= UTC_TIMESTAMP() - INTERVAL {sql_int(stale_lock_minutes)} MINUTE, 1, 0)),
|
|
'failed', SUM(IF(processed_at IS NULL AND attempts >= {sql_int(max_attempts)}, 1, 0)),
|
|
'processed', SUM(IF(processed_at IS NOT NULL, 1, 0)),
|
|
'oldest_pending_id', MIN(IF(processed_at IS NULL, id, NULL)),
|
|
'newest_pending_id', MAX(IF(processed_at IS NULL, id, NULL)),
|
|
'oldest_processed_at', DATE_FORMAT(MIN(processed_at), '%Y-%m-%dT%H:%i:%sZ'),
|
|
'newest_processed_at', DATE_FORMAT(MAX(processed_at), '%Y-%m-%dT%H:%i:%sZ')
|
|
) AS CHAR)) AS document
|
|
FROM event_outbox_events;
|
|
"""
|
|
)
|
|
status = rows[0] if rows else {}
|
|
return {key: (0 if value is None and key in {"total", "pending", "ready", "locked", "failed", "processed"} else value) for key, value in status.items()}
|
|
|
|
|
|
def claim_events(
|
|
remote,
|
|
worker_id,
|
|
limit,
|
|
max_attempts,
|
|
stale_lock_minutes,
|
|
):
|
|
remote.mysql(
|
|
f"""
|
|
UPDATE event_outbox_events
|
|
SET locked_at = UTC_TIMESTAMP(), locked_by = {sql_string(worker_id)}
|
|
WHERE processed_at IS NULL
|
|
AND attempts < {sql_int(max_attempts)}
|
|
AND (locked_at IS NULL OR locked_at < UTC_TIMESTAMP() - INTERVAL {sql_int(stale_lock_minutes)} MINUTE)
|
|
ORDER BY id
|
|
LIMIT {sql_int(limit)};
|
|
"""
|
|
)
|
|
return remote.mysql_json_lines(
|
|
f"""
|
|
SELECT HEX(CAST(JSON_OBJECT(
|
|
'id', id,
|
|
'event_type', event_type,
|
|
'source_type', source_type,
|
|
'source_id', source_id,
|
|
'project_id', project_id,
|
|
'issue_id', issue_id,
|
|
'journal_id', journal_id,
|
|
'user_id', user_id,
|
|
'occurred_at', DATE_FORMAT(occurred_at, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'attempts', attempts,
|
|
'payload', payload
|
|
) AS CHAR)) AS document
|
|
FROM event_outbox_events
|
|
WHERE processed_at IS NULL
|
|
AND locked_by = {sql_string(worker_id)}
|
|
ORDER BY id
|
|
LIMIT {sql_int(limit)};
|
|
"""
|
|
)
|
|
|
|
|
|
def purge_processed(remote, days, apply):
|
|
if days < 0:
|
|
raise OutboxWorkerError("--purge-processed-days must be zero or greater.")
|
|
count_sql = f"""
|
|
SELECT HEX(CAST(JSON_OBJECT(
|
|
'count', COUNT(*)
|
|
) AS CHAR)) AS document
|
|
FROM event_outbox_events
|
|
WHERE processed_at IS NOT NULL
|
|
AND processed_at < UTC_TIMESTAMP() - INTERVAL {sql_int(days)} DAY;
|
|
"""
|
|
rows = remote.mysql_json_lines(count_sql)
|
|
count = int((rows[0] if rows else {}).get("count") or 0)
|
|
if apply and count:
|
|
remote.mysql(
|
|
f"""
|
|
DELETE FROM event_outbox_events
|
|
WHERE processed_at IS NOT NULL
|
|
AND processed_at < UTC_TIMESTAMP() - INTERVAL {sql_int(days)} DAY;
|
|
"""
|
|
)
|
|
return count
|
|
|
|
|
|
def enrich_event(remote, event):
|
|
payload = parse_payload(event.get("payload"))
|
|
documents = [event_document(event, payload)]
|
|
event_type = str(event.get("event_type") or "")
|
|
|
|
if event_type.startswith("helpdesk_ticket."):
|
|
documents.extend(fetch_ticket_documents(remote, [payload.get("helpdesk_ticket_id") or event.get("source_id")]))
|
|
elif event_type.startswith("journal_message."):
|
|
documents.extend(fetch_message_documents(remote, [payload.get("journal_message_id") or event.get("source_id")]))
|
|
elif event_type.startswith("issue.") and event.get("issue_id"):
|
|
documents.extend(fetch_tickets_by_issue(remote, [event.get("issue_id")]))
|
|
elif event_type.startswith("journal.") and event.get("journal_id"):
|
|
documents.extend(fetch_messages_by_journal(remote, [event.get("journal_id")]))
|
|
elif event_type.startswith("contact."):
|
|
documents.extend(fetch_contact_documents(remote, [payload.get("contact_id") or event.get("source_id")]))
|
|
|
|
return [with_event_context(document, event) for document in documents]
|
|
|
|
|
|
def event_document(event, payload):
|
|
return {
|
|
"doc_type": "event",
|
|
"doc_id": f"event:{event.get('id')}",
|
|
"event_id": event.get("id"),
|
|
"event_type": event.get("event_type"),
|
|
"source_type": event.get("source_type"),
|
|
"source_id": event.get("source_id"),
|
|
"project_id": event.get("project_id"),
|
|
"issue_id": event.get("issue_id"),
|
|
"journal_id": event.get("journal_id"),
|
|
"user_id": event.get("user_id"),
|
|
"occurred_at": event.get("occurred_at"),
|
|
"payload": payload,
|
|
}
|
|
|
|
|
|
def fetch_ticket_documents(remote, ids):
|
|
id_list = sql_id_list(ids)
|
|
if not id_list:
|
|
return []
|
|
return remote.mysql_json_lines(ticket_sql(f"ht.id IN ({id_list})"))
|
|
|
|
|
|
def fetch_tickets_by_issue(remote, issue_ids):
|
|
id_list = sql_id_list(issue_ids)
|
|
if not id_list:
|
|
return []
|
|
return remote.mysql_json_lines(ticket_sql(f"ht.issue_id IN ({id_list})"))
|
|
|
|
|
|
def fetch_message_documents(remote, ids):
|
|
id_list = sql_id_list(ids)
|
|
if not id_list:
|
|
return []
|
|
return remote.mysql_json_lines(message_sql(f"jm.id IN ({id_list})"))
|
|
|
|
|
|
def fetch_messages_by_journal(remote, journal_ids):
|
|
id_list = sql_id_list(journal_ids)
|
|
if not id_list:
|
|
return []
|
|
return remote.mysql_json_lines(message_sql(f"jm.journal_id IN ({id_list})"))
|
|
|
|
|
|
def fetch_contact_documents(remote, ids):
|
|
id_list = sql_id_list(ids)
|
|
if not id_list:
|
|
return []
|
|
return remote.mysql_json_lines(
|
|
f"""
|
|
SELECT HEX(CAST(JSON_OBJECT(
|
|
'doc_type', 'contact',
|
|
'doc_id', CONCAT('contact:', c.id),
|
|
'contact_id', c.id,
|
|
'contact_name', TRIM(CONCAT_WS(' ', c.first_name, c.middle_name, c.last_name)),
|
|
'contact_company', c.company,
|
|
'contact_email', c.email,
|
|
'is_company', c.is_company,
|
|
'created_on', DATE_FORMAT(c.created_on, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'updated_on', DATE_FORMAT(c.updated_on, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'text', CONCAT_WS('\\n',
|
|
TRIM(CONCAT_WS(' ', c.first_name, c.middle_name, c.last_name)),
|
|
c.company,
|
|
c.email
|
|
)
|
|
) AS CHAR)) AS document
|
|
FROM contacts c
|
|
WHERE c.id IN ({id_list})
|
|
ORDER BY c.id;
|
|
"""
|
|
)
|
|
|
|
|
|
def ticket_sql(where_clause):
|
|
return f"""
|
|
SELECT HEX(CAST(JSON_OBJECT(
|
|
'doc_type', 'ticket',
|
|
'doc_id', CONCAT('ticket:', ht.id),
|
|
'helpdesk_ticket_id', ht.id,
|
|
'issue_id', i.id,
|
|
'project_id', i.project_id,
|
|
'project_identifier', p.identifier,
|
|
'contact_id', ht.contact_id,
|
|
'contact_name', TRIM(CONCAT_WS(' ', c.first_name, c.middle_name, c.last_name)),
|
|
'contact_company', c.company,
|
|
'contact_email', c.email,
|
|
'from_address', ht.from_address,
|
|
'to_address', ht.to_address,
|
|
'cc_address', ht.cc_address,
|
|
'message_id', ht.message_id,
|
|
'source', ht.source,
|
|
'is_incoming', ht.is_incoming,
|
|
'issue_subject', i.subject,
|
|
'status', s.name,
|
|
'tracker', t.name,
|
|
'assigned_to', TRIM(CONCAT_WS(' ', au.firstname, au.lastname)),
|
|
'ticket_date', DATE_FORMAT(ht.ticket_date, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'issue_updated_on', DATE_FORMAT(i.updated_on, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'text', CONCAT_WS('\\n',
|
|
i.subject,
|
|
LEFT(i.description, 8000),
|
|
TRIM(CONCAT_WS(' ', c.first_name, c.middle_name, c.last_name)),
|
|
c.company,
|
|
c.email,
|
|
ht.from_address,
|
|
ht.to_address,
|
|
ht.cc_address
|
|
)
|
|
) AS CHAR)) AS document
|
|
FROM helpdesk_tickets ht
|
|
JOIN issues i ON i.id = ht.issue_id
|
|
LEFT JOIN contacts c ON c.id = ht.contact_id
|
|
LEFT JOIN projects p ON p.id = i.project_id
|
|
LEFT JOIN issue_statuses s ON s.id = i.status_id
|
|
LEFT JOIN trackers t ON t.id = i.tracker_id
|
|
LEFT JOIN users au ON au.id = i.assigned_to_id
|
|
WHERE {where_clause}
|
|
ORDER BY ht.id;
|
|
"""
|
|
|
|
|
|
def message_sql(where_clause):
|
|
return f"""
|
|
SELECT HEX(CAST(JSON_OBJECT(
|
|
'doc_type', 'message',
|
|
'doc_id', CONCAT('message:', jm.id),
|
|
'journal_message_id', jm.id,
|
|
'journal_id', j.id,
|
|
'issue_id', i.id,
|
|
'project_id', i.project_id,
|
|
'project_identifier', p.identifier,
|
|
'contact_id', jm.contact_id,
|
|
'contact_name', TRIM(CONCAT_WS(' ', c.first_name, c.middle_name, c.last_name)),
|
|
'contact_company', c.company,
|
|
'contact_email', c.email,
|
|
'from_address', jm.from_address,
|
|
'to_address', jm.to_address,
|
|
'cc_address', jm.cc_address,
|
|
'has_bcc_address', IF(jm.bcc_address IS NULL OR jm.bcc_address = '', false, true),
|
|
'message_id', jm.message_id,
|
|
'source', jm.source,
|
|
'is_incoming', jm.is_incoming,
|
|
'issue_subject', i.subject,
|
|
'status', s.name,
|
|
'tracker', t.name,
|
|
'journal_user', TRIM(CONCAT_WS(' ', ju.firstname, ju.lastname)),
|
|
'message_date', DATE_FORMAT(jm.message_date, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'journal_created_on', DATE_FORMAT(j.created_on, '%Y-%m-%dT%H:%i:%sZ'),
|
|
'text', CONCAT_WS('\\n',
|
|
i.subject,
|
|
LEFT(j.notes, 8000),
|
|
TRIM(CONCAT_WS(' ', c.first_name, c.middle_name, c.last_name)),
|
|
c.company,
|
|
c.email,
|
|
jm.from_address,
|
|
jm.to_address,
|
|
jm.cc_address
|
|
)
|
|
) AS CHAR)) AS document
|
|
FROM journal_messages jm
|
|
JOIN journals j ON j.id = jm.journal_id
|
|
JOIN issues i ON i.id = j.journalized_id AND j.journalized_type = 'Issue'
|
|
LEFT JOIN contacts c ON c.id = jm.contact_id
|
|
LEFT JOIN projects p ON p.id = i.project_id
|
|
LEFT JOIN issue_statuses s ON s.id = i.status_id
|
|
LEFT JOIN trackers t ON t.id = i.tracker_id
|
|
LEFT JOIN users ju ON ju.id = j.user_id
|
|
WHERE {where_clause}
|
|
ORDER BY jm.id;
|
|
"""
|
|
|
|
|
|
def with_event_context(document, event):
|
|
document["event_id"] = event.get("id")
|
|
document["event_type"] = event.get("event_type")
|
|
document["event_occurred_at"] = event.get("occurred_at")
|
|
document["derived_at"] = int(time.time())
|
|
return document
|
|
|
|
|
|
def append_jsonl(path, documents):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a", encoding="utf-8") as handle:
|
|
for document in documents:
|
|
handle.write(json.dumps(document, ensure_ascii=False, sort_keys=True))
|
|
handle.write("\n")
|
|
|
|
|
|
def mark_processed(remote, event_id, worker_id):
|
|
remote.mysql(
|
|
f"""
|
|
UPDATE event_outbox_events
|
|
SET processed_at = UTC_TIMESTAMP(), locked_at = NULL, locked_by = NULL, last_error = NULL
|
|
WHERE id = {sql_int(event_id)}
|
|
AND locked_by = {sql_string(worker_id)};
|
|
"""
|
|
)
|
|
|
|
|
|
def mark_failed(remote, event_id, worker_id, exc):
|
|
message = f"{exc.__class__.__name__}: {exc}"
|
|
remote.mysql(
|
|
f"""
|
|
UPDATE event_outbox_events
|
|
SET attempts = attempts + 1,
|
|
last_error = {sql_string(message[:4000])},
|
|
locked_at = NULL,
|
|
locked_by = NULL
|
|
WHERE id = {sql_int(event_id)}
|
|
AND locked_by = {sql_string(worker_id)};
|
|
"""
|
|
)
|
|
|
|
|
|
def release_claims(remote, worker_id):
|
|
remote.mysql(
|
|
f"""
|
|
UPDATE event_outbox_events
|
|
SET locked_at = NULL, locked_by = NULL
|
|
WHERE processed_at IS NULL
|
|
AND locked_by = {sql_string(worker_id)};
|
|
"""
|
|
)
|
|
|
|
|
|
def parse_payload(value):
|
|
if isinstance(value, dict):
|
|
return value
|
|
if not value:
|
|
return {}
|
|
try:
|
|
parsed = json.loads(str(value))
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
|
|
|
|
def sql_id_list(values):
|
|
ids = []
|
|
for value in values:
|
|
try:
|
|
int_value = int(value)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if int_value > 0:
|
|
ids.append(str(int_value))
|
|
return ",".join(sorted(set(ids), key=int))
|
|
|
|
|
|
def sql_int(value):
|
|
try:
|
|
return max(0, int(value))
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def sql_string(value):
|
|
return "'" + str(value).replace("\\", "\\\\").replace("'", "\\'") + "'"
|
|
|
|
|
|
def shell_quote(value):
|
|
return "'" + value.replace("'", "'\"'\"'") + "'"
|
|
|
|
|
|
def make_worker_id():
|
|
return f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:12]}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|