Pretvarjanje dokumentov DOCX v Markdown s Pythonom
Python je eden najbolj priljubljenih jezikov za obdelavo dokumentov — in z dobrim razlogom. Njegov ekosistem knjižnic za obdelavo datotek, asinhrono mreženje in obdelavo podatkov ga naredi idealnega za naloge paketne pretvorbe.
Ta vadnica vas bo vodila skozi uporabo API-ja AnyMD iz Pythona za pretvorbo datotek DOCX v čisti Markdown, vključno z asinhrono obdelavo, obravnavo napak in paginacijo.
Zahteve
Potrebovali boste:
- Python 3.9+
httpxknjižnico (za asinhrone zahtevke)- Veljaven API ključ AnyMD (pridobite ga v nadzorni plošči)
Namestite odvisnosti:
pip install httpx
Osnovna pretvorba: DOCX v Markdown
Najprej preprost primer — naložite datoteko DOCX in prejmite Markdown:
import httpx
api_key = "md-..."
file_path = "document.docx"
with open(file_path, "rb") as f:
response = httpx.post(
"https://anymd.net/api/convert",
headers={"Authorization": f"Bearer {api_key}"},
files={"file": f},
)
response.raise_for_status()
data = response.json()
print(data["markdown"])
# "## Poglavje 1\n\nTo je prvi odstavek..."
To je vse, kar potrebujete za pretvorbo ene datoteke. API vrže {"markdown": "..."} s čistim Markdownom.
Paketna pretvorba z asinhrono obdelavo
Ko morate pretvoriti desetine ali stotine datotek DOCX, želite asinhrono pošiljanje, da se izognete zaporednemu čakanju:
import asyncio
import httpx
from pathlib import Path
API_KEY = "md-..."
API_URL = "https://anymd.net/api/convert"
async def convert_file(client, path: Path) -> dict:
with open(path, "rb") as f:
response = await client.post(
API_URL,
headers={"Authorization": f"Bearer {API_KEY}"},
files={"file": f},
)
response.raise_for_status()
return {"file": path.name, **response.json()}
async def batch_convert(file_paths: list[Path]) -> list[dict]:
limits = httpx.Limits(max_keepalive_connections=10, max_connections=10)
async with httpx.AsyncClient(limits=limits) as client:
tasks = [convert_file(client, p) for p in file_paths]
results = await asyncio.gather(*tasks, return_exceptions=True)
successes, errors = [], []
for path, result in zip(file_paths, results):
if isinstance(result, Exception):
errors.append({"file": path.name, "error": str(result)})
else:
successes.append(result)
return {"successes": successes, "errors": errors}
# Uporaba
files = list(Path("docs/").glob("*.docx"))
results = asyncio.run(batch_convert(files))
print(f"Pretvorjeno: {len(results['successes'])} datotek")
print(f"Napake: {len(results['errors'])} datotek")
Ta vzorec uporablja asyncio.gather za hkratno obdelavo do 10 datotek. Za večje serije (1000+ datotek) priporočamo dodajanje nadzora nad sočasnostjo s semaforjem:
sem = asyncio.Semaphore(5)
async def convert_file(client, path):
async with sem:
# ... koda za pretvorbo
Obravnava napak in ponovni poskusi
Omrežne napake in občasne težave s strežnikom so neizogibne. Robustna rešitev vključuje ponovne poskuse z eksponentnim zamikom:
import time
import random
def convert_with_retry(file_path, max_retries=3):
for attempt in range(max_retries):
try:
with open(file_path, "rb") as f:
response = httpx.post(
"https://anymd.net/api/convert",
headers={"Authorization": f"Bearer {API_KEY}"},
files={"file": f},
timeout=30.0,
)
response.raise_for_status()
return response.json()
except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
if attempt == max_retries - 1:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
print(f"Poskus {attempt + 1} ni uspel, čakanje {wait:.1f}s...")
time.sleep(wait)
# Uporaba
result = convert_with_retry("document.docx")
Streaming velikih datotek
Za velike datoteke DOCX (100 MB+) lahko odgovor streamate, da se izognete preobremenitvi pomnilnika:
with open("large.docx", "rb") as f:
with httpx.Client() as client:
with client.stream(
"POST",
"https://anymd.net/api/convert",
headers={"Authorization": f"Bearer {API_KEY}"},
files={"file": f},
) as response:
response.raise_for_status()
for chunk in response.iter_bytes():
# Obdelujte dele Markdown sproti
process_chunk(chunk)
Celoten cevovod: DOCX → Markdown → RAG
Ko pretvorite v Markdown, lahko nadaljujete z razdelitvijo (chunkingom) in indeksiranjem v RAG cevovodu:
import httpx
from langchain.text_splitter import MarkdownHeaderTextSplitter
API_KEY = "md-..."
# 1. Pretvori DOCX v Markdown
with open("report.docx", "rb") as f:
resp = httpx.post(
"https://anymd.net/api/convert",
headers={"Authorization": f"Bearer {API_KEY}"},
files={"file": f},
)
markdown = resp.json()["markdown"]
# 2. Razdeli Markdown na podlagi naslovov
headers_to_split_on = [
("#", "Naslov 1"),
("##", "Naslov 2"),
("###", "Naslov 3"),
]
splitter = MarkdownHeaderTextSplitter(headers_to_split_on)
chunks = splitter.split_text(markdown)
# 3. Indeksiraj v vektorski bazi
for chunk in chunks:
embedding = get_embedding(chunk.page_content)
vector_store.add(embedding, metadata=chunk.metadata)
Meritve zmogljivosti
| Velikost paketa | Sočasnost | Skupni čas | Povprečno na datoteko |
|---|---|---|---|
| 10 datotek | 1 (zaporedno) | ~8 s | ~800 ms |
| 50 datotek | 10 (hkrati) | ~4 s | ~80 ms |
| 100 datotek | 10 (hkrati) | ~8 s | ~80 ms |
| 500 datotek | 10 (hkrati) | ~40 s | ~80 ms |
| 1000 datotek | 10 (hkrati) | ~80 s | ~80 ms |
Pretvorba je skoraj linearno razširljiva s sočasnostjo. Za res velike serije razmislite o uporabi ukazne vrstice ali orodij za paketno obdelavo.
Naslednji koraki
- Pretvori dokumente iz ukazne vrstice ali Pythona — popoln pregled vseh načinov uporabe
- Dokumentacija API-ja — vse končne točke in parametri
- Cenik — brezplačen preizkus, plačilo po porabi