File size: 2,876 Bytes
a6d7bb2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | # xbrl_processing/instance_finder.py
import tempfile
import zipfile
import requests
import io
from xbrl_processing.downloader import download_xbrl
def file_contains_ixbrl(text: str) -> bool:
if not text:
return False
t = text.lower()
return (
"<ix:" in t or
"<ix:non" in t or
"<ix:header" in t or
"xmlns:ix" in t
)
def file_contains_xbrl_xml(text: str) -> bool:
if not text:
return False
t = text.lower()
return (
"<xbrli:" in t or
"xmlns:xbrli" in t or
"http://www.xbrl.org" in t
)
def find_esef_xhtml_in_zip(url: str):
try:
resp = requests.get(url, timeout=15)
z = zipfile.ZipFile(io.BytesIO(resp.content))
candidates = [
name for name in z.namelist()
if name.lower().endswith((".xhtml", ".html"))
]
if not candidates:
return None, None
for name in candidates:
try:
content = z.read(name).decode("utf-8", errors="ignore")
if file_contains_ixbrl(content):
return z, name
except:
pass
for name in candidates:
if "report" in name.lower():
return z, name
if "xbrl" in name.lower():
return z, name
largest = max(candidates, key=lambda n: len(z.read(n)))
return z, largest
except Exception:
return None, None
def find_valid_instance(df):
"""
Returns: (instance_local_path, instance_source_url)
"""
# ZIP → XHTML
zip_rows = df[df["Url"].str.contains(".zip", case=False, na=False)]
if not zip_rows.empty:
for _, row in zip_rows.iterrows():
zfile, entry = find_esef_xhtml_in_zip(row["Url"])
if entry:
with tempfile.NamedTemporaryFile(suffix=".xhtml", delete=False) as tmp:
tmp.write(zfile.read(entry))
tmp.flush()
return tmp.name, row["Url"]
# XML → XBRL
xml_rows = df[
df["Url"].str.contains(".xml", case=False, na=False) |
df["Filtype"].str.contains("XBRL", case=False, na=False) |
df["Url"].str.contains("xbrl", case=False, na=False)
]
if not xml_rows.empty:
for _, row in xml_rows.iterrows():
try:
resp = requests.get(row["Url"], timeout=10)
chunk = resp.content[:200000].decode("utf-8", errors="ignore")
if file_contains_xbrl_xml(chunk):
with tempfile.NamedTemporaryFile(suffix=".xml", delete=False) as tmp:
download_xbrl(row["Url"], tmp.name)
return tmp.name, row["Url"]
except Exception:
pass
# Nothing found
return None, None
|