File size: 2,934 Bytes
b156b8a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import os
import requests
import tempfile
import pandas as pd
import PyPDF2
from tools.interfaces.interface_file_reader import FileReaderInterface

class HuggingFaceFileReader(FileReaderInterface):
    
    def __init__(self, huggingface_token: str, base_url: str,is_local=False):
        self.huggingface_token = huggingface_token
        self.base_url = base_url
        self.is_local=is_local

    def read(self, file_path: str) -> str:
        try:
            # Si le fichier est local, on lit directement
            if self.is_local:
                if os.path.isfile(file_path):
                    return self._read_local(file_path)

            # Sinon, on considère que c'est un fichier distant
            full_url = self.base_url + file_path
            headers = {"Authorization": f"Bearer {self.huggingface_token}"}

            if file_path.lower().endswith(".txt") or file_path.lower().endswith(".pdf"):
                # Télécharger et lire localement
                with tempfile.NamedTemporaryFile(delete=False, suffix=os.path.splitext(file_path)[-1]) as tmp_file:
                    response = requests.get(full_url, headers=headers)
                    response.raise_for_status()
                    tmp_file.write(response.content)
                    tmp_path = tmp_file.name

                content = self._read_local(tmp_path)
                os.remove(tmp_path)
                return content

            elif file_path.lower().endswith(".csv"):
                response = requests.get(full_url, headers=headers)
                response.raise_for_status()
                df = pd.read_csv(pd.compat.StringIO(response.text))
                return df.to_string(index=False)

            elif file_path.lower().endswith((".xls", ".xlsx")):
                response = requests.get(full_url, headers=headers)
                response.raise_for_status()
                with tempfile.NamedTemporaryFile(suffix=".xlsx") as tmp:
                    tmp.write(response.content)
                    tmp.flush()
                    df = pd.read_excel(tmp.name)
                return df.to_string(index=False)

            else:
                return "Format de fichier non supporté. Utilise .txt, .csv, .xlsx ou .pdf"

        except Exception as e:
            return f"Erreur lors de la lecture du fichier : {e}"

    def _read_local(self, path: str) -> str:
        if path.lower().endswith(".txt"):
            with open(path, "r", encoding="utf-8") as f:
                return f.read()
        elif path.lower().endswith(".pdf"):
            text = ""
            with open(path, "rb") as f:
                reader = PyPDF2.PdfReader(f)
                for page in reader.pages:
                    text += page.extract_text() or ""
            return text if text.strip() else "Aucun texte lisible extrait du PDF."
        else:
            return "Format local non supporté pour cette méthode."