3 Commits

8 changed files with 27 additions and 330 deletions

2
.gitignore vendored
View File

@@ -205,5 +205,3 @@ cython_debug/
marimo/_static/
marimo/_lsp/
__marimo__/
*.csv

175
src/scraper.py → main.py Executable file → Normal file
View File

@@ -1,23 +1,14 @@
#!/usr/bin/env python3
from sys import argv
from typing import cast
from requests import HTTPError, Response, Session
from requests import Response, Session
from bs4 import BeautifulSoup, Tag
from collections import OrderedDict
from json import JSONDecodeError, loads
from json import loads
class _ScraperData:
"""_summary_"""
def __init__(self, data: dict[str, object]) -> None:
"""_summary_
Args:
data (dict[str, object]): _description_
"""
def __init__(self, data: dict[str, object], scraper: 'Scraper | None' = None) -> None:
self._data: dict[str, object] = data
self._scraper: Scraper | None = scraper
def _getcontent(self) -> dict[str, object] | None:
"""_summary_
@@ -45,51 +36,6 @@ class _ScraperData:
return None
return cast(dict[str, object], current_data.get("attributes"))
def prix(self) -> float | None:
"""
Retourne le prix unitaire d'une bouteille (75cl).
Si aucun prix n'est disponible, retourne None.
"""
content = self._getcontent()
if content is None:
return None
items = content.get("items")
# Vérifie que items existe et n'est pas vide
if not isinstance(items, list) or len(items) == 0:
return None
prix_calcule: float | None = None
for item in items:
if not isinstance(item, dict):
continue
p = item.get("offerPrice")
attrs = item.get("attributes", {})
nbunit = attrs.get("nbunit", {}).get("value")
equivbtl = attrs.get("equivbtl", {}).get("value")
if not isinstance(p, (int, float)) or not nbunit or not equivbtl:
continue
nb = float(nbunit)
eq = float(equivbtl)
if nb <= 0 or eq <= 0:
continue
if nb == 1 and eq == 1:
return float(p)
prix_calcule = round(float(p) / (nb * eq), 2)
return prix_calcule
def appellation(self) -> str | None:
"""_summary_
@@ -123,8 +69,8 @@ class _ScraperData:
return None
val = cast(str, app_dict.get("value")).rstrip("+").split("-")
if len(val) > 1 and val[1] != "":
val[0] = str(round((float(val[0]) + float(val[1])) / 2, 1))
if len(val) > 1:
val[0] = str((int(val[0]) + int(val[1])) / 2)
return val[0]
return None
@@ -141,23 +87,6 @@ class _ScraperData:
def getdata(self) -> dict[str, object]:
return self._data
def informations(self) -> str:
"""
Retourne toutes les informations sous la forme :
"Appelation,Parker,J.Robinson,J.Suckling,Prix"
"""
appellation = self.appellation()
parker = self.parker()
robinson = self.robinson()
suckling = self.suckling()
try:
prix = self.prix()
except ValueError:
prix = None
return f"{appellation},{parker},{robinson},{suckling},{prix}"
class Scraper:
"""
@@ -174,18 +103,6 @@ class Scraper:
# Très utile pour éviter de renvoyer toujours les mêmes handshake
# TCP et d'avoir toujours une connexion constante avec le server
self._session: Session = Session()
# Crée une "fausse carte d'identité" pour éviter que le site nous
# bloque car on serait des robots
self._session.headers.update(
{
"User-Agent":
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) \
AppleWebKit/537.36 (KHTML, like Gecko) \
Chrome/122.0.0.0 Safari/537.36",
"Accept-Language":
"fr-FR,fr;q=0.9,en;q=0.8",
}
)
# Système de cache pour éviter de solliciter le serveur inutilement
self._latest_request: tuple[(str, Response)] | None = None
self._latest_soups: OrderedDict[str, BeautifulSoup] = OrderedDict[
@@ -206,8 +123,7 @@ class Scraper:
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
"""
target_url: str = self._url + subdir.lstrip("/")
# envoyer une requête GET sur la page si erreur, renvoie un raise
response: Response = self._session.get(url=target_url, timeout=30)
response: Response = self._session.get(url=target_url, timeout=10)
response.raise_for_status()
return response
@@ -306,80 +222,3 @@ class Scraper:
raise ValueError(f"Clé manquante dans le JSON : {key}")
return _ScraperData(cast(dict[str, object], current_data))
def _geturlproductslist(self, subdir: str) -> list[str] | None:
"""_summary_
Args:
subdir (str): _description_
Returns:
_type_: _description_
"""
try:
data: dict[str, object] = self.getjsondata(subdir).getdata()
for element in ["initialReduxState", "categ", "content"]:
data: dict[str, object] = cast(dict[str, object], data.get(element))
if not isinstance(data, dict):
return None
products: list[str] = cast(list[str], data.get("products"))
if isinstance(products, list):
return products
except (JSONDecodeError, HTTPError):
return None
def getvins(self, subdir: str, filename: str) -> None:
"""_summary_
Args:
subdir (str): _description_
filename (str): _description_
"""
with open(filename, "w") as f:
cache: set[str] = set[str]()
page = 0
_ = f.write("Appellation,Robert,Robinson,Suckling,Prix\n")
while True:
page += 1
products_list: list[str] | None = \
self._geturlproductslist(f"{subdir}?page={page}")
if not products_list:
break
products_list_length = len(products_list)
for i, product in enumerate(products_list):
if not isinstance(product, dict):
continue
link = product.get("seoKeyword")
if link and link not in cache:
try:
infos = self.getjsondata(link).informations()
_ = f.write(infos + "\n")
print(
f"page: {page} | {i + 1}/{products_list_length} {link}"
)
cache.add(link)
except (JSONDecodeError, HTTPError) as e:
print(f"Erreur sur le produit {link}: {e}")
f.flush()
def main() -> None:
if len(argv) != 2:
raise ValueError(f"{argv[0]} <sous-url>")
scraper: Scraper = Scraper()
scraper.getvins(argv[1], "donnee.csv")
if __name__ == "__main__":
try:
main()
except Exception as e:
print(f"ERREUR: {e}")

View File

@@ -1,15 +0,0 @@
[project]
name = "projet-millesima-s6"
version = "0.1.0"
dependencies = [
"requests==2.32.5",
"beautifulsoup4==4.14.3",
"pandas==2.3.3",
]
[project.optional-dependencies]
test = ["pytest==8.4.2", "requests-mock==1.12.1"]
[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"

4
requirements.txt Normal file
View File

@@ -0,0 +1,4 @@
requests>=2.32.5
requests-mock>=1.12.1
beautifulsoup4>=4.14.3

Binary file not shown.

View File

@@ -1,20 +0,0 @@
#!/usr/bin/env python3
from os import getcwd
from os.path import normpath, join
from sys import argv
from pandas import read_csv, DataFrame
def main() -> None:
if len(argv) != 2:
raise ValueError(f"{argv[0]} <filename.csv>")
path: str = normpath(join(getcwd(), argv[1]))
db: DataFrame = read_csv(path)
print(db.all())
if __name__ == "__main__":
try:
main()
except Exception as e:
print(f"ERREUR: {e}")

View File

@@ -1,8 +1,8 @@
from json import dumps
from unittest.mock import patch, mock_open
from bs4 import Tag
import pytest
from requests_mock import Mocker
from scraper import Scraper
from main import Scraper
@pytest.fixture(autouse=True)
@@ -71,10 +71,10 @@ def mock_site():
"_id": "J4131/22/C/CC/6-11652",
"partnumber": "J4131/22/C/CC/6",
"taxRate": "H",
"listPrice": 842,
"offerPrice": 842,
"seoKeyword": "vin-de-charazade1867.html",
"shortdesc": "Une bouteille du meilleur vin du monde?",
"listPrice": 390,
"offerPrice": 390,
"seoKeyword": "nino-negri-5-stelle-sfursat-2022-c-cc-6.html",
"shortdesc": "Un carton de 6 Bouteilles (75cl)",
"attributes": {
"promotion_o_n": {
"valueId": "0",
@@ -94,18 +94,6 @@ def mock_site():
"type": "CHECKBOX",
"isSpirit": False,
},
"equivbtl": {
"valueId": "1",
"name": "equivbtl",
"value": "1",
"isSpirit": False,
},
"nbunit": {
"valueId": "1",
"name": "nbunit",
"value": "1",
"isSpirit": False,
},
},
"stock": 12,
"availability": "2026-02-05",
@@ -120,14 +108,14 @@ def mock_site():
"appellation": {
"valueId": "433",
"name": "Appellation",
"value": "Madame-Loïk",
"url": "Madame-loik.html",
"value": "Sforzato di Valtellina",
"url": "sforzato-di-valtellina.html",
"isSpirit": False,
"groupIdentifier": "appellation_433",
},
"note_rp": {
"valueId": "91",
"name": "Peter Parker",
"name": "Parker",
"value": "91",
"isSpirit": False,
},
@@ -138,8 +126,8 @@ def mock_site():
"isSpirit": False,
},
"note_js": {
"valueId": "93-94.5",
"name": "J. cherazade",
"valueId": "93-94",
"name": "J. Suckling",
"value": "93-94",
"isSpirit": False,
},
@@ -166,79 +154,6 @@ def mock_site():
text=html_product,
)
html_product = f"""
<html>
<body>
<h1>MILLESIMA</h1>
<script id="__NEXT_DATA__" type="application/json">
{dumps(json_data)}
</script>
</body>
</html>
"""
list_pleine = f"""
<html>
<body>
<h1>LE WINE</h1>
<script id="__NEXT_DATA__" type="application/json">
{dumps({
"props": {
"pageProps": {
"initialReduxState": {
"categ": {
"content": {
"products": [
{"seoKeyword": "/nino-negri-5-stelle-sfursat-2022.html",},
{"seoKeyword": "/poubelle",},
{"seoKeyword": "/",}
]
}
}
}
}
}
}
)}
</script>
</body>
</html>
"""
list_vide = f"""
<html>
<body>
<h1>LE WINE</h1>
<script id="__NEXT_DATA__" type="application/json">
{dumps({
"props": {
"pageProps": {
"initialReduxState": {
"categ": {
"content": {
"products": [
]
}
}
}
}
}
}
)}
</script>
</body>
</html>
"""
m.get(
"https://www.millesima.fr/wine.html",
complete_qs=False,
response_list=[
{"text": list_pleine},
{"text": list_vide},
],
)
# on return m sans fermer le server qui simule la page
yield m
@@ -248,6 +163,7 @@ def scraper() -> Scraper:
return Scraper()
# EXO1
def test_soup(scraper: Scraper):
vide = scraper.getsoup("")
poubelle = scraper.getsoup("poubelle")
@@ -257,15 +173,17 @@ def test_soup(scraper: Scraper):
assert str(contenu.find("h1")) == "<h1>MILLESIMA</h1>"
# EXO3
def test_appellation(scraper: Scraper):
vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle")
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
assert vide.appellation() is None
assert poubelle.appellation() is None
assert contenu.appellation() == "Madame-Loïk"
assert contenu.appellation() == "Sforzato di Valtellina"
# test fonctions privée
def test_fonctionprivee(scraper: Scraper):
vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle")
@@ -280,6 +198,7 @@ def test_fonctionprivee(scraper: Scraper):
assert contenu._getattributes() is not None
# EXO4-5
def test_critiques(scraper: Scraper):
vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle")
@@ -296,31 +215,3 @@ def test_critiques(scraper: Scraper):
assert contenu.robinson() == "17"
assert contenu.suckling() == "93.5"
assert contenu._getcritiques("test_ts") is None
def test_prix(scraper: Scraper):
vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle")
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
assert vide.prix() is None
assert poubelle.prix() is None
assert contenu.prix() == 842.0
def test_informations(scraper: Scraper):
contenu = scraper.getjsondata("nino-negri-5-stelle-sfursat-2022.html")
assert contenu.informations() == "Madame-Loïk,91,17,93.5,842.0"
vide = scraper.getjsondata("")
poubelle = scraper.getjsondata("poubelle")
assert vide.informations() == "None,None,None,None,None"
assert poubelle.informations() == "None,None,None,None,None"
def test_search(scraper: Scraper):
m = mock_open()
with patch("builtins.open", m):
scraper.getvins("wine.html", "fake_file.csv")
assert m().write.called
all_writes = "".join(call.args[0] for call in m().write.call_args_list)
assert "Madame-Loïk,91,17,93.5,842.0" in all_writes

View File