التعامل مع تنسيقات الملفات: CSV وJSON وTXT

أكثر ثلاثة تنسيقات شيوعًا لملفات الوكلاء

import csv
import json
from pathlib import Path

# Detect format from extension
def read_data_file(file_path):
    path = Path(file_path)
    if path.suffix == '.csv':
        return read_csv(path)
    elif path.suffix == '.json':
        return read_json(path)
    elif path.suffix == '.txt':
        return path.read_text(encoding='utf-8')
    else:
        raise ValueError(f'Unsupported format: {path.suffix}')

csv.reader — تحليل CSV الأساسي

import csv

with open('sales.csv', 'w', newline='') as f:
    f.write('product,qty,price\nWidget,3,9.99\nGadget,1,19.99\n')

with open('sales.csv', 'r', encoding='utf-8', newline='') as f:
    reader = csv.reader(f)
    header = next(reader)
    print('Columns:', header)

    for row in reader:
        product = row[0]
        quantity = int(row[1])
        price = float(row[2])
        print(f'{product}: {quantity} units at ${price}')

csv.DictReader — الصف في صورة قاموس

import csv

with open('employees.csv', 'w', newline='') as f:
    f.write('name,department,salary\nAlice,Eng,95000\nBob,Sales,70000\n')

with open('employees.csv', 'r', encoding='utf-8', newline='') as f:
    reader = csv.DictReader(f)
    print('Fields:', reader.fieldnames)

    total_salary = 0
    for row in reader:
        name = row['name']
        department = row['department']
        salary = float(row['salary'])
        total_salary += salary
        print(f'{name} ({department}): ${salary:,.2f}')

    print(f'Total payroll: ${total_salary:,.2f}')

csv.writer وDictWriter — كتابة CSV

import csv

results = [
    {'task_id': 'T001', 'status': 'completed', 'duration_s': 12.5},
    {'task_id': 'T002', 'status': 'failed', 'duration_s': 3.1},
    {'task_id': 'T003', 'status': 'completed', 'duration_s': 45.8},
]

fieldnames = ['task_id', 'status', 'duration_s']

with open('task_results.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()  # write column names
    writer.writerows(results)

print('Wrote task_results.csv')

json.load() وjson.dump() — إدخال وإخراج الملفات

import json

with open('config.json', 'w', encoding='utf-8') as f:
    json.dump({'api_url': 'https://api.example.com', 'timeout': 15}, f)

with open('config.json', 'r', encoding='utf-8') as f:
    config = json.load(f)

print('API URL:', config.get('api_url'))
print('Timeout:', config.get('timeout', 30))

output_data = {
    'run_id': 'abc123',
    'items': [1, 2, 3],
    'meta': {'agent': 'v2', 'model': 'gpt-4o'}
}

with open('output.json', 'w', encoding='utf-8') as f:
    json.dump(
        output_data, f,
        indent=2,
        ensure_ascii=False
    )
print('Written output.json')

معالجة أخطاء فك ترميز JSON

import json
from pathlib import Path

def safe_load_json(file_path):
    path = Path(file_path)
    try:
        with open(path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except json.JSONDecodeError as e:
        print(f'Invalid JSON in {path}: line {e.lineno}, col {e.colno}')
        print(f'  Error: {e.msg}')
        # Show the problem area
        content = path.read_text(encoding='utf-8')
        lines = content.split('\n')
        if e.lineno <= len(lines):
            print(f'  Content: {lines[e.lineno-1][:80]}')
        return None
    except FileNotFoundError:
        print(f'File not found: {path}')
        return None

# --- demo ---
Path('good.json').write_text('{"model": "gpt-4o"}', encoding='utf-8')
Path('bad.json').write_text('{"model": "gpt-4o", }', encoding='utf-8')

print('Loading good.json:', safe_load_json('good.json'))
print('Loading bad.json:', safe_load_json('bad.json'))

قراءة تنسيق JSONL (JSON Lines)

import json

with open('events.jsonl', 'w', encoding='utf-8') as f:
    f.write('{"event": "start"}\n{"event": "stop"}\nnot json\n')

results = []
with open('events.jsonl', 'r', encoding='utf-8') as f:
    for line_num, line in enumerate(f, 1):
        line = line.strip()
        if not line:
            continue
        try:
            event = json.loads(line)
            results.append(event)
        except json.JSONDecodeError as e:
            print(f'Bad JSON on line {line_num}: {e}')

print(f'Loaded {len(results)} events')

with open('output.jsonl', 'w', encoding='utf-8') as f:
    for record in results:
        f.write(json.dumps(record, ensure_ascii=False) + '\n')

قراءة ملفات النص العادي

from pathlib import Path

Path('system_prompt.txt').write_text('You are a helpful agent.', encoding='utf-8')
with open('agent.log', 'w', encoding='utf-8') as f:
    f.write('INFO: boot\nERROR: disk full\nCRITICAL: crash\nINFO: recovered\n')

prompt = Path('system_prompt.txt').read_text(encoding='utf-8')
print(f'Prompt length: {len(prompt)} characters')

error_lines = []
with open('agent.log', 'r', encoding='utf-8') as f:
    for line in f:
        line = line.rstrip()
        if not line:
            continue
        if 'ERROR' in line or 'CRITICAL' in line:
            error_lines.append(line)

print(f'Found {len(error_lines)} error lines')

with open('summary.txt', 'w', encoding='utf-8') as f:
    f.write('Agent Run Summary\n')
    f.write('=' * 40 + '\n')
    for error in error_lines[:10]:
        f.write(f'  {error}\n')
print('Summary written')

معالجة BOM (علامة ترتيب البايتات)

import csv

with open('windows_export.csv', 'wb') as f:
    f.write('name,email,age\nAlice,alice@x.com,30\n'.encode('utf-8'))
with open('file.txt', 'wb') as f:
    f.write(b'\xef\xbb\xbfhello')

with open('windows_export.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    first = next(reader)
    print(list(first.keys()))

with open('windows_export.csv', 'r', encoding='utf-8-sig') as f:
    reader = csv.DictReader(f)
    first = next(reader)
    print(list(first.keys()))

content = open('file.txt', 'rb').read()
if content.startswith(b'\xef\xbb\xbf'):
    content = content[3:]
text = content.decode('utf-8')
print('Decoded:', text)

معالجة أخطاء الترميز

from pathlib import Path

def read_with_fallback(file_path):
    path = Path(file_path)

    # Try UTF-8 first
    try:
        return path.read_text(encoding='utf-8')
    except UnicodeDecodeError:
        pass

    # Try Latin-1 (handles most European files)
    try:
        return path.read_text(encoding='latin-1')
    except UnicodeDecodeError:
        pass

    # Last resort: replace bad characters
    text = path.read_text(encoding='utf-8', errors='replace')
    print(f'Warning: {path.name} had encoding errors (chars replaced)')
    return text

# --- demo ---
Path('notes_utf8.txt').write_text('Notes: café, naïve, résumé', encoding='utf-8')
text = read_with_fallback('notes_utf8.txt')
print(f'Read {len(text)} chars: {text!r}')

معالجة ملفات CSV غير السليمة

import csv

with open('messy_data.csv', 'w', newline='') as f:
    f.write('name,email,score\nAlice,alice@x.com,88\nBob,,90\nCarol,carol@x.com,notanumber\n')

valid_rows = []
error_count = 0

with open('messy_data.csv', 'r', encoding='utf-8', newline='') as f:
    reader = csv.DictReader(f)
    expected_fields = {'name', 'email', 'score'}

    for line_num, row in enumerate(reader, start=2):
        try:
            if not all(row.get(f, '').strip() for f in expected_fields):
                raise ValueError(f'Missing required field in row {line_num}')
            score = float(row['score'])
            valid_rows.append({
                'name': row['name'].strip(),
                'email': row['email'].strip().lower(),
                'score': score
            })
        except (ValueError, KeyError) as e:
            error_count += 1
            print(f'Skipping row {line_num}: {e}')

print(f'Valid: {len(valid_rows)}, Errors: {error_count}')

اختبار سريع: CSV.DictReader مقابل reader

مراجعة معالجة تنسيقات الملفات