Claude Files API — upload, reference, or retention failure

import os
import mimetypes
import anthropic
from pathlib import Path

client = anthropic.Anthropic(
    default_headers={"anthropic-beta": "files-api-2025-04-14"}  # verify current beta
)

LIMITS = {
    "application/pdf": 32 * 1024 * 1024,
    "text/plain":      32 * 1024 * 1024,
    "image/png":       5  * 1024 * 1024,
    "image/jpeg":      5  * 1024 * 1024,
    "image/gif":       5  * 1024 * 1024,
    "image/webp":      5  * 1024 * 1024,
}

def upload_file(path: str, purpose: str = "user_data") -> str:
    """Validate then upload. Returns file_id."""
    p = Path(path)
    if not p.is_file():
        raise FileNotFoundError(path)

    mime, _ = mimetypes.guess_type(str(p))
    if mime not in LIMITS:
        raise ValueError(
            f"Unsupported MIME {mime!r} for {path}. Convert to PDF or text first."
        )

    size = p.stat().st_size
    if size > LIMITS[mime]:
        raise ValueError(
            f"{path} is {size/1024/1024:.1f} MB — over {LIMITS[mime]/1024/1024:.0f} MB "
            f"limit for {mime}. Split or compress."
        )

    with p.open("rb") as f:
        uploaded = client.beta.files.upload(file=(p.name, f, mime))

    return uploaded.id

file_id = upload_file("report.pdf")

# Reference the uploaded file in a Messages call
response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=1000,
    messages=[{
        "role": "user",
        "content": [
            {"type": "document", "source": {"type": "file", "file_id": file_id}},
            {"type": "text", "text": "Summarise the key financial risks."},
        ],
    }],
)
print(response.content[0].text)

from pypdf import PdfReader, PdfWriter
from pathlib import Path

def split_pdf(source: str, out_dir: str = "chunks", pages_per_chunk: int = 30) -> list[str]:
    reader = PdfReader(source)
    Path(out_dir).mkdir(exist_ok=True)
    chunks = []
    for start in range(0, len(reader.pages), pages_per_chunk):
        writer = PdfWriter()
        for i in range(start, min(start + pages_per_chunk, len(reader.pages))):
            writer.add_page(reader.pages[i])
        chunk_path = f"{out_dir}/chunk_{start//pages_per_chunk:03d}.pdf"
        with open(chunk_path, "wb") as f:
            writer.write(f)
        chunks.append(chunk_path)
    return chunks

# Upload each chunk, keep order
chunks = split_pdf("giant_report.pdf", pages_per_chunk=30)
file_ids = [upload_file(c) for c in chunks]

# Reference all chunks in one Messages call — Claude reads them as a sequence
content = [
    {"type": "document", "source": {"type": "file", "file_id": fid}}
    for fid in file_ids
] + [{"type": "text", "text": "Summarise findings across all sections."}]

response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=2000,
    messages=[{"role": "user", "content": content}],
)

import anthropic
import hashlib
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Optional

client = anthropic.Anthropic(
    default_headers={"anthropic-beta": "files-api-2025-04-14"}
)

# Your file record table
@dataclass
class FileRecord:
    logical_id: str        # your app's ID
    file_id: str           # Anthropic's ID
    source_path: str
    sha256: str
    uploaded_at: datetime
    ttl_days: int = 30

    @property
    def expires_at(self) -> datetime:
        return self.uploaded_at + timedelta(days=self.ttl_days)

    @property
    def expiring_soon(self) -> bool:
        return self.expires_at - datetime.utcnow() < timedelta(days=3)

def compute_sha256(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()

def refresh_if_expiring(record: FileRecord) -> FileRecord:
    if not record.expiring_soon:
        return record

    # Old file will disappear soon — re-upload
    new_file_id = upload_file(record.source_path)  # from previous fix

    # Delete the old one to keep the workspace clean
    try:
        client.beta.files.delete(record.file_id)
    except anthropic.NotFoundError:
        pass  # already gone

    return FileRecord(
        logical_id=record.logical_id,
        file_id=new_file_id,
        source_path=record.source_path,
        sha256=record.sha256,
        uploaded_at=datetime.utcnow(),
    )

# Nightly job — refresh anything expiring in the next 3 days
for record in all_file_records():
    updated = refresh_if_expiring(record)
    if updated.file_id != record.file_id:
        save_record(updated)