|
import os |
|
import sys |
|
import json |
|
import shutil |
|
import subprocess |
|
import tempfile |
|
from datetime import datetime |
|
from xml.etree import ElementTree as ET |
|
|
|
import requests |
|
from requests.auth import HTTPBasicAuth |
|
from pypdfium2 import PdfDocument |
|
import openpyxl |
|
from markitdown import MarkItDown |
|
|
|
|
|
CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bcard_crm_config.json") |
|
WEBDAV_BASE = "https://your-koofr-instance.net/dav" |
|
CARD_FOLDER = "/path/to/business/cards" |
|
CRM_FILE = "/path/to/crm.xlsx" |
|
CRM_SHEET = "Potentials (Entry)" |
|
OCR_TIMEOUT = 30 |
|
|
|
NS = {"d": "DAV:"} |
|
|
|
|
|
def load_koofr_creds(path=None): |
|
if path and os.path.exists(path): |
|
user = None |
|
pw = None |
|
with open(path) as f: |
|
data = json.load(f) |
|
user = data.get("koofr_username") |
|
pw = data.get("koofr_password") |
|
if user and pw: |
|
return user, pw |
|
|
|
if os.path.exists(CONFIG_FILE): |
|
with open(CONFIG_FILE) as f: |
|
data = json.load(f) |
|
user = data.get("koofr_username") |
|
pw = data.get("koofr_password") |
|
if user and pw: |
|
return user, pw |
|
|
|
raise RuntimeError("No Koofr credentials found. Create bcard_crm_config.json next to this script.") |
|
|
|
|
|
def webdav_auth(): |
|
user, pw = load_koofr_creds() |
|
return HTTPBasicAuth(user, pw) |
|
|
|
|
|
def list_dir(path, auth=None): |
|
if auth is None: |
|
auth = webdav_auth() |
|
url = WEBDAV_BASE + path |
|
headers = {"Depth": "1"} |
|
r = requests.request("PROPFIND", url, auth=auth, headers=headers) |
|
r.raise_for_status() |
|
tree = ET.fromstring(r.content) |
|
results = [] |
|
for resp in tree.findall(".//d:response", NS): |
|
href = resp.find("d:href", NS).text |
|
raw = href.rstrip("/").split("/")[-1] |
|
decoded = requests.utils.unquote(raw) |
|
if decoded and decoded != path.strip("/").split("/")[-1]: |
|
results.append(decoded) |
|
return results |
|
|
|
|
|
def download_file(remote_path, local_path, auth=None): |
|
if auth is None: |
|
auth = webdav_auth() |
|
url = WEBDAV_BASE + remote_path |
|
r = requests.get(url, auth=auth, stream=True) |
|
r.raise_for_status() |
|
with open(local_path, "wb") as f: |
|
for chunk in r.iter_content(8192): |
|
f.write(chunk) |
|
return r.status_code |
|
|
|
|
|
def upload_file(local_path, remote_path, auth=None): |
|
if auth is None: |
|
auth = webdav_auth() |
|
url = WEBDAV_BASE + remote_path |
|
with open(local_path, "rb") as f: |
|
r = requests.put(url, auth=auth, data=f) |
|
r.raise_for_status() |
|
return r.status_code |
|
|
|
|
|
def pdf_to_png(pdf_path, output_path, scale=2): |
|
doc = PdfDocument(pdf_path) |
|
page = doc[0] |
|
bitmap = page.render(scale=scale) |
|
bitmap.to_pil().save(output_path) |
|
|
|
|
|
def ocr_localai(image_path): |
|
cli = "/path/to/localai_cli.py" |
|
try: |
|
result = subprocess.run( |
|
[sys.executable, cli, "ocr", "-i", image_path], |
|
capture_output=True, text=True, timeout=OCR_TIMEOUT |
|
) |
|
if result.returncode == 0 and result.stdout.strip(): |
|
return result.stdout.strip(), "localai" |
|
except subprocess.TimeoutExpired: |
|
pass |
|
return None, "localai-timeout" |
|
|
|
|
|
def ocr_markitdown(pdf_path): |
|
md = MarkItDown() |
|
try: |
|
result = md.convert(pdf_path) |
|
if result.text_content and result.text_content.strip(): |
|
return result.text_content.strip(), "markitdown" |
|
except Exception: |
|
pass |
|
return None, "markitdown-error" |
|
|
|
|
|
def ocr_card(pdf_path): |
|
tmpdir = tempfile.mkdtemp() |
|
png_path = os.path.join(tmpdir, "page.png") |
|
try: |
|
pdf_to_png(pdf_path, png_path) |
|
text, source = ocr_localai(png_path) |
|
if text: |
|
return text, source |
|
text, source = ocr_markitdown(pdf_path) |
|
if text: |
|
return text, source |
|
raise RuntimeError("Both OCR methods failed") |
|
finally: |
|
shutil.rmtree(tmpdir, ignore_errors=True) |
|
|
|
|
|
def cmd_list(args): |
|
folder = args.folder if args.folder else CARD_FOLDER |
|
subfolder = "" |
|
if args.event: |
|
subfolder = "/" + args.event |
|
path = folder + subfolder |
|
items = list_dir(path) |
|
pdfs = [i for i in items if i.lower().endswith(".pdf")] |
|
for p in sorted(pdfs): |
|
print(p) |
|
|
|
|
|
def cmd_list_folders(args): |
|
items = list_dir(CARD_FOLDER) |
|
for i in sorted(items): |
|
print(i) |
|
|
|
|
|
def cmd_ocr(args): |
|
remote = CARD_FOLDER + "/" |
|
if args.event: |
|
remote += args.event + "/" |
|
remote += args.file |
|
tmpdir = tempfile.mkdtemp() |
|
local_pdf = os.path.join(tmpdir, args.file) |
|
try: |
|
download_file(remote, local_pdf) |
|
text, source = ocr_card(local_pdf) |
|
print(f"[OCR via {source}]") |
|
print(text) |
|
finally: |
|
shutil.rmtree(tmpdir, ignore_errors=True) |
|
|
|
|
|
HEADERS = [ |
|
"ID", "Date Created", "Company Name", "Contact Name", |
|
"Contact Role", "Contact Info", "Market / Industry", |
|
"Company Type", "Focus Area", "Source", |
|
"Operational Region", "Additional Regions", "Internal Lead", |
|
"Potential Status", "Notes", |
|
] |
|
|
|
HEADER_ROW = 11 |
|
|
|
|
|
def read_crm_headers(): |
|
tmpdir = tempfile.mkdtemp() |
|
local = os.path.join(tmpdir, "crm.xlsx") |
|
try: |
|
download_file(CRM_FILE, local) |
|
wb = openpyxl.load_workbook(local, read_only=True) |
|
ws = wb[CRM_SHEET] |
|
headers = {} |
|
for cell in ws[HEADER_ROW]: |
|
if cell.value: |
|
headers[cell.column_letter] = cell.value |
|
wb.close() |
|
return headers |
|
finally: |
|
shutil.rmtree(tmpdir, ignore_errors=True) |
|
|
|
|
|
def create_export(output_path, rows): |
|
wb = openpyxl.Workbook() |
|
ws = wb.active |
|
ws.title = CRM_SHEET |
|
|
|
ws["A1"] = "Business Card Leads Export" |
|
ws["A1"].font = openpyxl.styles.Font(bold=True, size=14) |
|
|
|
ws["A2"] = f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}" |
|
ws["A2"].font = openpyxl.styles.Font(italic=True, size=10, color="808080") |
|
|
|
for col_idx, header in enumerate(HEADERS, 3): |
|
cell = ws.cell(row=HEADER_ROW, column=col_idx, value=header) |
|
cell.font = openpyxl.styles.Font(bold=True) |
|
cell.fill = openpyxl.styles.PatternFill(start_color="D9E2F3", end_color="D9E2F3", fill_type="solid") |
|
|
|
key_order = [ |
|
"company_name", "contact_name", "contact_role", "contact_info", |
|
"market_industry", "company_type", "focus_area", "source", |
|
"operational_region", "additional_regions", "internal_lead", |
|
"potential_status", "notes", |
|
] |
|
|
|
for row_idx, row_data in enumerate(rows, HEADER_ROW + 1): |
|
for col_idx, key in enumerate(key_order, 5): |
|
ws.cell(row=row_idx, column=col_idx, value=row_data.get(key, "")) |
|
|
|
for col_idx in range(3, 3 + len(HEADERS)): |
|
col_letter = openpyxl.utils.get_column_letter(col_idx) |
|
ws.column_dimensions[col_letter].width = 20 |
|
|
|
ws.column_dimensions["G"].width = 30 |
|
ws.column_dimensions["H"].width = 35 |
|
ws.column_dimensions["Q"].width = 40 |
|
|
|
wb.save(output_path) |
|
wb.close() |
|
|
|
|
|
def cmd_export(args): |
|
rows = [] |
|
for json_file in args.files: |
|
with open(json_file) as f: |
|
rows.append(json.load(f)) |
|
if not rows: |
|
print("No row files provided.") |
|
return |
|
|
|
output = args.output if args.output else "/tmp/business_card_leads.xlsx" |
|
create_export(output, rows) |
|
print(f"Exported {len(rows)} leads to {output}") |
|
|
|
|
|
def cmd_export_append(args): |
|
output = args.output if args.output else "/tmp/business_card_leads.xlsx" |
|
row_data = json.loads(args.json) |
|
|
|
if os.path.exists(output): |
|
wb = openpyxl.load_workbook(output) |
|
ws = wb.active |
|
next_row = ws.max_row + 1 |
|
key_order = [ |
|
"company_name", "contact_name", "contact_role", "contact_info", |
|
"market_industry", "company_type", "focus_area", "source", |
|
"operational_region", "additional_regions", "internal_lead", |
|
"potential_status", "notes", |
|
] |
|
for col_idx, key in enumerate(key_order, 5): |
|
ws.cell(row=next_row, column=col_idx, value=row_data.get(key, "")) |
|
wb.save(output) |
|
wb.close() |
|
print(f"Appended to {output}, row {next_row}") |
|
else: |
|
create_export(output, [row_data]) |
|
print(f"Created {output} with first row") |
|
|
|
|
|
def cmd_upload(args): |
|
local = args.input if args.input else "/tmp/business_card_leads.xlsx" |
|
remote = args.remote if args.remote else CARD_FOLDER + "/business card leads.xlsx" |
|
if not os.path.exists(local): |
|
raise RuntimeError(f"File not found: {local}") |
|
upload_file(local, remote) |
|
print(f"Uploaded {local} -> {remote}") |
|
|
|
|
|
def cmd_process(args): |
|
folder = CARD_FOLDER |
|
if args.event: |
|
folder = folder + "/" + args.event |
|
|
|
items = list_dir(folder) |
|
pdfs = sorted([i for i in items if i.lower().endswith(".pdf")]) |
|
|
|
if not pdfs: |
|
print("No PDF files found in folder.") |
|
return |
|
|
|
for pdf_name in pdfs: |
|
tmpdir = tempfile.mkdtemp() |
|
local_pdf = os.path.join(tmpdir, pdf_name) |
|
try: |
|
download_file(folder + "/" + pdf_name, local_pdf) |
|
text, source = ocr_card(local_pdf) |
|
print(f"=== {pdf_name} [{source}] ===") |
|
print(text) |
|
print() |
|
finally: |
|
shutil.rmtree(tmpdir, ignore_errors=True) |
|
|
|
|
|
def main(): |
|
import argparse |
|
parser = argparse.ArgumentParser(description="Business card OCR + CRM tool") |
|
sub = parser.add_subparsers(dest="command", required=True) |
|
|
|
p = sub.add_parser("list", help="List business card PDFs in a folder") |
|
p.add_argument("--event", help="Subfolder/event name") |
|
p.add_argument("--folder", help="Override base folder path") |
|
|
|
sub.add_parser("list-folders", help="List subfolders in business cards") |
|
|
|
p = sub.add_parser("ocr", help="OCR a single business card PDF") |
|
p.add_argument("file", help="PDF filename") |
|
p.add_argument("--event", help="Subfolder/event name") |
|
|
|
p = sub.add_parser("process", help="OCR all PDFs in a folder") |
|
p.add_argument("--event", help="Subfolder/event name") |
|
|
|
p = sub.add_parser("export", help="Create leads export Excel from JSON row files") |
|
p.add_argument("files", nargs="+", help="JSON files with row data") |
|
p.add_argument("--output", help="Output Excel path") |
|
|
|
p = sub.add_parser("export-append", help="Append a row to the export Excel") |
|
p.add_argument("--json", required=True, help="Row data as JSON") |
|
p.add_argument("--output", help="Excel file path") |
|
|
|
p = sub.add_parser("upload", help="Upload a file to Koofr") |
|
p.add_argument("--input", help="Local file path") |
|
p.add_argument("--remote", help="Remote Koofr path") |
|
|
|
args = parser.parse_args() |
|
|
|
commands = { |
|
"list": cmd_list, |
|
"list-folders": cmd_list_folders, |
|
"ocr": cmd_ocr, |
|
"process": cmd_process, |
|
"export": cmd_export, |
|
"export-append": cmd_export_append, |
|
"upload": cmd_upload, |
|
} |
|
commands[args.command](args) |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |