9 Commits

12 changed files with 270 additions and 287 deletions

View File

@@ -5,35 +5,41 @@ name: Python application
on: on:
push: push:
branches: [ "main" ] branches: ["main"]
pull_request: pull_request:
branches: [ "main" ] branches: ["main"]
permissions: permissions:
contents: read contents: write
jobs: jobs:
build: build:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v4
- name: Set up Python 3.10
uses: actions/setup-python@v3 - name: Set up Python 3.10
with: uses: actions/setup-python@v4
python-version: "3.10" with:
- name: Install dependencies python-version: "3.10"
run: |
python -m pip install --upgrade pip - name: install dependencies
pip install flake8 pytest run: |
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi python -m pip install --upgrade pip
- name: Lint with flake8 pip install ".[test,doc]"
run: |
# stop the build if there are Python syntax errors or undefined names - name: Lint with flake8
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics run: |
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
run: | - name: Test with pytest
pytest run: pytest
- name: Deploy Doc
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
run: |
git config user.name github-actions
git config user.email github-actions@github.com
mkdocs gh-deploy --force

View File

@@ -1,48 +0,0 @@
#!/usr/bin/env python3
from pandas import DataFrame, to_numeric
def display_info(df: DataFrame) -> None:
print(df.all())
print(df.info())
print("\nNombre de valeurs manquantes par colonne :")
print(df.isna().sum())
def drop_empty_appellation(df: DataFrame) -> DataFrame:
return df.dropna(subset=["Appellation"])
def mean_score(df: DataFrame, col: str) -> DataFrame:
"""
Calcule la moyenne d'une colonne de score par appellation.
- Convertit les valeurs en numériques, en remplaçant les non-convertibles par NaN
- Calcule la moyenne par appellation
- Remplace les NaN résultants par 0
"""
tmp = df[["Appellation", col]].copy()
tmp[col] = to_numeric(tmp[col], errors="coerce")
# moyenne par appellation
means = tmp.groupby("Appellation", as_index=False)[col].mean()
means[col] = means[col].fillna(0)
means = means.rename(columns={col: f"mean_{col}"})
return means
def mean_robert(df: DataFrame) -> DataFrame:
return mean_score(df, "Robert")
def mean_robinson(df: DataFrame) -> DataFrame:
return mean_score(df, "Robinson")
def mean_suckling(df: DataFrame) -> DataFrame:
return mean_score(df, "Suckling")

1
docs/index.md Normal file
View File

@@ -0,0 +1 @@
# Millesima

3
docs/scraper.md Normal file
View File

@@ -0,0 +1,3 @@
# Scraper
::: scraper.Scraper

4
docs/scraperdata.md Normal file
View File

@@ -0,0 +1,4 @@
# _ScraperData
::: scraper._ScraperData

60
main.py
View File

@@ -1,60 +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
from cleaning import (display_info,
drop_empty_appellation,
mean_robert,
mean_robinson,
mean_suckling)
def load_csv(filename: str) -> DataFrame:
path: str = normpath(join(getcwd(), filename))
return read_csv(path)
def save_csv(df: DataFrame, out_filename: str) -> None:
df.to_csv(out_filename, index=False)
def main() -> None:
if len(argv) != 2:
raise ValueError(f"Usage: {argv[0]} <filename.csv>")
df = load_csv(argv[1])
print("=== Avant nettoyage ===")
display_info(df)
df = drop_empty_appellation(df)
save_csv(df, "donnee_clean.csv")
print("\n=== Après nettoyage d'appellations manquantes ===")
display_info(df)
#la moyenne des notes des vins pour chaque appellation
robert_means = mean_robert(df)
save_csv(robert_means, "mean_robert_by_appellation.csv")
print("\n=== moyenne Robert par appellation ===")
print(robert_means.head(10))
robinson_means = mean_robinson(df)
save_csv(robinson_means, "mean_robinson_by_appellation.csv")
print("\n===: moyennes Robinson par appellation ===")
print(robinson_means.head(10))
suckling_means = mean_suckling(df)
save_csv(suckling_means, "mean_suckling_by_appellation.csv")
print("\n===: moyennes Suckling par appellation ===")
print(suckling_means.head(10))
if __name__ == "__main__":
try:
main()
except Exception as e:
print(f"ERREUR: {e}")

14
mkdocs.yml Normal file
View File

@@ -0,0 +1,14 @@
site_name: "Projet Millesima S6"
theme:
name: "material"
plugins:
- search
- mkdocstrings
markdown_extensions:
- admonition
- pymdownx.details
- pymdownx.superfences
- pymdownx.tabbed

12
pyproject.toml Normal file
View File

@@ -0,0 +1,12 @@
[project]
name = "projet-millesima-s6"
version = "0.1.0"
dependencies = ["requests==2.32.5", "beautifulsoup4==4.14.3", "pandas==2.3.3", "tqdm==4.67.3"]
[project.optional-dependencies]
test = ["pytest==8.4.2", "requests-mock==1.12.1", "flake8==7.3.0"]
doc = ["mkdocs<2.0.0", "mkdocs-material==9.6.23", "mkdocstrings[python]"]
[build-system]
requires = ["setuptools", "wheel"]
build-backend = "setuptools.build_meta"

View File

@@ -1,6 +0,0 @@
requests==2.32.5
requests-mock==1.12.1
beautifulsoup4==4.14.3
pytest==8.4.2
requests-mock==1.12.1
pandas==2.3.3

View File

@@ -1,32 +1,78 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
from sys import argv
from typing import cast
from requests import HTTPError, Response, Session
from requests.exceptions import Timeout, ConnectionError
import time
from bs4 import BeautifulSoup, Tag
from collections import OrderedDict from collections import OrderedDict
from io import SEEK_END, SEEK_SET, BufferedWriter
from json import JSONDecodeError, loads from json import JSONDecodeError, loads
from pathlib import Path from os import makedirs
from os.path import dirname, exists, join, normpath, realpath
from pickle import UnpicklingError, dump, load
from sys import argv
from tqdm.std import tqdm
from typing import Any, Callable, Literal, TypeVar, cast
from bs4 import BeautifulSoup, Tag
from requests import HTTPError, Response, Session
_dir: str = dirname(realpath(__name__))
T = TypeVar("T")
def _getcache(mode: Literal["rb", "wb"], fn: Callable[[Any], T]) -> T | None:
"""_summary_
Returns:
_type_: _description_
"""
cache_dirname = normpath(join(_dir, ".cache"))
save_path = normpath(join(cache_dirname, "save"))
if not exists(cache_dirname):
makedirs(cache_dirname)
try:
with open(save_path, mode) as f:
return fn(f)
except (FileNotFoundError, EOFError, UnpicklingError):
return None
def savestate(data: tuple[int, set[str]]) -> None:
def save(f: BufferedWriter) -> None:
_ = f.seek(0)
_ = f.truncate()
dump(data, f)
f.flush()
_getcache("wb", save)
def loadstate() -> tuple[int, set[str]] | None:
return _getcache("rb", lambda f: load(f))
class _ScraperData: class _ScraperData:
"""_summary_""" """
Conteneur de données spécialisé pour extraire les informations des dictionnaires JSON.
Cette classe agit comme une interface simplifiée au-dessus du dictionnaire brut
renvoyé par la balise __NEXT_DATA__ du site Millesima.
"""
def __init__(self, data: dict[str, object]) -> None: def __init__(self, data: dict[str, object]) -> None:
"""_summary_ """
Initialise le conteneur avec un dictionnaire de données.
Args: Args:
data (dict[str, object]): _description_ data (dict[str, object]): Le dictionnaire JSON brut extrait de la page.
""" """
self._data: dict[str, object] = data self._data: dict[str, object] = data
def _getcontent(self) -> dict[str, object] | None: def _getcontent(self) -> dict[str, object] | None:
"""_summary_ """
Navigue dans l'arborescence Redux pour atteindre le contenu du produit.
Returns: Returns:
dict[str, object]: _description_ dict[str, object] | None: Le dictionnaire du produit ou None si la structure diffère.
""" """
current_data: dict[str, object] = self._data current_data: dict[str, object] = self._data
for key in ["initialReduxState", "product", "content"]: for key in ["initialReduxState", "product", "content"]:
@@ -38,10 +84,11 @@ class _ScraperData:
return current_data return current_data
def _getattributes(self) -> dict[str, object] | None: def _getattributes(self) -> dict[str, object] | None:
"""_summary_ """
Extrait les attributs techniques (notes, appellations, etc.) du produit.
Returns: Returns:
dict[str, object]: _description_ dict[str, object] | None: Les attributs du vin ou None.
""" """
current_data: object = self._getcontent() current_data: object = self._getcontent()
if current_data is None: if current_data is None:
@@ -50,9 +97,13 @@ class _ScraperData:
def prix(self) -> float | None: def prix(self) -> float | None:
""" """
Retourne le prix unitaire d'une bouteille (75cl). Calcule le prix unitaire d'une bouteille (standardisée à 75cl).
Si aucun prix n'est disponible, retourne None. Le site vend souvent par caisses (6, 12 bouteilles) ou formats (Magnum).
Cette méthode normalise le prix pour obtenir celui d'une seule unité.
Returns:
float | None: Le prix calculé arrondi à 2 décimales, ou None.
""" """
content = self._getcontent() content = self._getcontent()
@@ -94,13 +145,13 @@ class _ScraperData:
return prix_calcule return prix_calcule
def appellation(self) -> str | None: def appellation(self) -> str | None:
"""_summary_ """
Extrait le nom de l'appellation du vin.
Returns: Returns:
str: _description_ str | None: Le nom (ex: 'Pauillac') ou None.
""" """
attrs: dict[str, object] | None = self._getattributes() attrs: dict[str, object] | None = self._getattributes()
if attrs is not None: if attrs is not None:
app_dict: object | None = attrs.get("appellation") app_dict: object | None = attrs.get("appellation")
if isinstance(app_dict, dict): if isinstance(app_dict, dict):
@@ -108,13 +159,16 @@ class _ScraperData:
return None return None
def _getcritiques(self, name: str) -> str | None: def _getcritiques(self, name: str) -> str | None:
"""_summary_ """
Méthode générique pour parser les notes des critiques (Parker, Suckling, etc.).
Gère les notes simples ("95") et les plages de notes ("95-97") en faisant la moyenne.
Args: Args:
name (str): _description_ name (str): La clé de l'attribut dans le JSON (ex: 'note_rp').
Returns: Returns:
str | None: _description_ str | None: La note formatée en chaîne de caractères ou None.
""" """
current_value: dict[str, object] | None = self._getattributes() current_value: dict[str, object] | None = self._getattributes()
@@ -133,21 +187,27 @@ class _ScraperData:
return None return None
def parker(self) -> str | None: def parker(self) -> str | None:
"""Note Robert Parker."""
return self._getcritiques("note_rp") return self._getcritiques("note_rp")
def robinson(self) -> str | None: def robinson(self) -> str | None:
"""Note Jancis Robinson."""
return self._getcritiques("note_jr") return self._getcritiques("note_jr")
def suckling(self) -> str | None: def suckling(self) -> str | None:
"""Note James Suckling."""
return self._getcritiques("note_js") return self._getcritiques("note_js")
def getdata(self) -> dict[str, object]: def getdata(self) -> dict[str, object]:
"""Retourne le dictionnaire de données complet."""
return self._data return self._data
def informations(self) -> str: def informations(self) -> str:
""" """
Retourne toutes les informations sous la forme : Agrège les données clés pour l'export CSV.
"Appelation,Parker,J.Robinson,J.Suckling,Prix"
Returns:
str: Ligne formatée : "Appellation,Parker,Robinson,Suckling,Prix".
""" """
appellation = self.appellation() appellation = self.appellation()
@@ -161,27 +221,38 @@ class _ScraperData:
class Scraper: class Scraper:
""" """
Scraper est une classe qui permet de gerer Client HTTP optimisé pour le scraping de millesima.fr.
de façon dynamique des requetes uniquement
sur le serveur https de Millesima Gère la session persistante, les headers de navigation et un cache double
pour optimiser les performances et la discrétion.
""" """
def __init__(self) -> None: def __init__(self) -> None:
""" """
Initialise la session de scraping. Initialise l'infrastructure de navigation:
- créer une session pour éviter de faire un handshake pour chaque requête
- ajout d'un header pour éviter le blocage de l'accès au site
- ajout d'un système de cache
""" """
self._url: str = "https://www.millesima.fr/" self._url: str = "https://www.millesima.fr/"
# Très utile pour éviter de renvoyer toujours les mêmes handshake # Très utile pour éviter de renvoyer toujours les mêmes handshake
# TCP et d'avoir toujours une connexion constante avec le server # TCP et d'avoir toujours une connexion constante avec le server
self._session: Session = Session() self._session: Session = Session()
self._session.headers.update({ # Crée une "fausse carte d'identité" pour éviter que le site nous
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " # bloque car on serait des robots
"AppleWebKit/537.36 (KHTML, like Gecko) " self._session.headers.update(
"Chrome/122.0.0.0 Safari/537.36", {
"Accept-Language": "fr-FR,fr;q=0.9,en;q=0.8", "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 # Système de cache pour éviter de solliciter le serveur inutilement
# utilise pour _request
self._latest_request: tuple[(str, Response)] | None = None self._latest_request: tuple[(str, Response)] | None = None
# utilise pour getsoup
self._latest_soups: OrderedDict[str, BeautifulSoup] = OrderedDict[ self._latest_soups: OrderedDict[str, BeautifulSoup] = OrderedDict[
str, BeautifulSoup str, BeautifulSoup
]() ]()
@@ -196,24 +267,14 @@ class Scraper:
Returns: Returns:
Response: L'objet réponse de la requête. Response: L'objet réponse de la requête.
Raise: Raises:
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx). HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
""" """
target_url: str = self._url + subdir.lstrip("/") target_url: str = self._url + subdir.lstrip("/")
# envoyer une requête GET sur la page si erreur, renvoie un raise
last_exc: Exception | None = None response: Response = self._session.get(url=target_url, timeout=30)
for attempt in range(1, 4): response.raise_for_status()
try: return response
response: Response = self._session.get(url=target_url, timeout=30)
response.raise_for_status()
return response
except (Timeout, ConnectionError) as e:
last_exc = e
print(f"Timeout/ConnectionError ({attempt}/3) sur {target_url}: {e}")
time.sleep(2 * attempt) # 2s, 4s, 6s
# après 3 essais, on abandonne
raise last_exc if last_exc else RuntimeError("Request failed")
def getresponse(self, subdir: str = "", use_cache: bool = True) -> Response: def getresponse(self, subdir: str = "", use_cache: bool = True) -> Response:
""" """
@@ -227,7 +288,7 @@ class Scraper:
Returns: Returns:
Response: L'objet réponse (cache ou nouvelle requête). Response: L'objet réponse (cache ou nouvelle requête).
Raise: Raises:
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx). HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
""" """
@@ -257,7 +318,7 @@ class Scraper:
Returns: Returns:
BeautifulSoup: L'objet parsé pour extraction de données. BeautifulSoup: L'objet parsé pour extraction de données.
Raise: Raises:
HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx). HTTPError: Si le serveur renvoie un code d'erreur (4xx, 5xx).
""" """
@@ -278,23 +339,20 @@ class Scraper:
def getjsondata(self, subdir: str, id: str = "__NEXT_DATA__") -> _ScraperData: def getjsondata(self, subdir: str, id: str = "__NEXT_DATA__") -> _ScraperData:
""" """
Extrait les données JSON contenues dans la balise __NEXT_DATA__ du site. Extrait les données JSON contenues dans la balise __NEXT_DATA__ du site.
Beaucoup de sites modernes (Next.js) stockent leur état initial dans
une balise <script> pour l'hydratation côté client.
Args: Args:
subdir (str): Le chemin de la page. subdir (str): Le chemin de la page.
id (str, optional): L'identifiant de la balise script (par défaut __NEXT_DATA__). id (str, optional): L'identifiant de la balise script.
Raises: Raises:
HTTPError: Soulevée par `getresponse` si le serveur renvoie un code d'erreur (4xx, 5xx). HTTPError: Erreur renvoyée par le serveur (4xx, 5xx).
JSONDecodeError: Soulevée par `loads` si le contenu de la balise n'est pas un JSON valide. JSONDecodeError: Si le contenu de la balise n'est pas un JSON valide.
ValueError: Soulevée manuellement si l'une des clés attendues (props, pageProps, etc.) ValueError: Si les clés 'props' ou 'pageProps' sont absentes.
est absente de la structure JSON.
Returns: Returns:
dict[str, object]: Un dictionnaire contenant les données utiles _ScraperData: Instance contenant les données extraites.
ou un dictionnaire vide en cas d'erreur.
""" """
soup: BeautifulSoup = self.getsoup(subdir) soup: BeautifulSoup = self.getsoup(subdir)
script: Tag | None = soup.find("script", id=id) script: Tag | None = soup.find("script", id=id)
@@ -311,117 +369,116 @@ class Scraper:
return _ScraperData(cast(dict[str, object], current_data)) return _ScraperData(cast(dict[str, object], current_data))
def _geturlproductslist(self, subdir: str): def _geturlproductslist(self, subdir: str) -> list[dict[str, Any]] | None:
"""_summary_ """
Récupère la liste des produits d'une page de catégorie.
Args:
subdir (str): _description_
Returns:
_type_: _description_
""" """
try: try:
data: dict[str, object] = self.getjsondata(subdir).getdata() data: dict[str, object] = self.getjsondata(subdir).getdata()
for element in ["initialReduxState", "categ", "content"]: for element in ["initialReduxState", "categ", "content"]:
data: dict[str, object] = cast(dict[str, object], data.get(element)) data = cast(dict[str, object], data.get(element))
if not isinstance(data, dict):
return None products: list[dict[str, Any]] = cast(
list[dict[str, Any]], data.get("products")
)
products: list[str] = cast(list[str], data.get("products"))
if isinstance(products, list): if isinstance(products, list):
return products return products
except (JSONDecodeError, HTTPError): except (JSONDecodeError, HTTPError):
return None return None
def _save_progress(self, page: int, i: int, last_link: str) -> None: def _writevins(self, cache: set[str], product: dict[str, Any], f: Any) -> None:
Path("progress.txt").write_text(f"{page},{i},{last_link}", encoding="utf-8")
def _load_progress(self) -> tuple[int, int, str | None]:
p = Path("progress.txt")
if not p.exists():
return (1, 0, None)
try:
parts = p.read_text(encoding="utf-8").strip().split(",", 2)
page = int(parts[0])
i = int(parts[1])
last_link = parts[2] if len(parts) == 3 and parts[2] != "" else None
return (page, i, last_link)
except Exception:
return (1, 0, None)
def getvins(self, subdir: str, filename: str):
"""_summary_ """_summary_
Args: Args:
subdir (str): _description_ cache (set[str]): _description_
filename (str): _description_ product (dict): _description_
f (Any): _description_
""" """
start_page, start_i, last_link = self._load_progress() if isinstance(product, dict):
print(f"__INFO__ Reprise à page={start_page}, index={start_i}, last_link={last_link}") link: Any | None = product.get("seoKeyword")
if link and link not in cache:
try:
infos = self.getjsondata(link).informations()
_ = f.write(infos + "\n")
cache.add(link)
except (JSONDecodeError, HTTPError) as e:
print(f"Erreur sur le produit {link}: {e}")
with open(filename, "a", encoding="utf-8") as f: def getvins(self, subdir: str, filename: str, reset: bool = False) -> None:
cache: set[str] = set[str]() """
Scrape toutes les pages d'une catégorie et sauvegarde en CSV.
if f.tell() == 0: Args:
_ = f.write("Appellation,Robert,Robinson,Suckling,Prix\n") subdir (str): La catégorie (ex: '/vins-rouges').
filename (str): Nom du fichier de sortie (ex: 'vins.csv').
reset (bool): (Optionnel) pour réinitialiser le processus.
"""
# mode d'écriture fichier
mode: Literal["w", "a+"] = "w" if reset else "a+"
# titre
title: str = "Appellation,Robert,Robinson,Suckling,Prix\n"
# page du début
page: int = 1
# le set qui sert de cache
cache: set[str] = set[str]()
page = start_page - 1 custom_format = "{l_bar} {bar:20} {r_bar}"
while True: if not reset:
page += 1 # appelle la fonction pour load le cache, si il existe
products_list = self._geturlproductslist(f"{subdir}?page={page}") # pas, il utilise les variables de base sinon il override
# toute les variables pour continuer et pas recommencer le
# processus en entier.
serializable: tuple[int, set[str]] | None = loadstate()
if isinstance(serializable, tuple):
# override la page et le cache
page, cache = serializable
try:
with open(filename, mode) as f:
# check si le titre est bien présent au début du buffer
# sinon il l'ecrit, petit bug potentiel, a+ ecrit tout le
# temps a la fin du buffer, si on a ecrit des choses avant
# le titre sera apres ces données mais on part du principe
# que personne va toucher le fichier.
_ = f.seek(0, SEEK_SET)
if not (f.read(len(title)) == title):
_ = f.write(title)
else:
_ = f.seek(0, SEEK_END)
if not products_list: while True:
break products_list: list[dict[str, Any]] | None = (
self._geturlproductslist(f"{subdir}?page={page}")
)
if not products_list:
break
products_list_length = len(products_list) pbar: tqdm[dict[str, Any]] = tqdm(
start_at = start_i if page == start_page else 0 products_list, bar_format=custom_format
)
for i in range(start_at, products_list_length): for product in pbar:
product = products_list[i] keyword = product.get("seoKeyword", "Inconnu")[:40]
if not isinstance(product, dict): pbar.set_description(
continue f"Page: {page:<3} | Product: {keyword:<40}"
)
link = product.get("seoKeyword") self._writevins(cache, product, f)
if not link: page += 1
continue except (Exception, HTTPError, KeyboardInterrupt, JSONDecodeError):
if not reset:
# pour eviter les doublons : savestate((page, cache))
if (page == start_page) and (last_link is not None) and (link == last_link):
self._save_progress(page, + 1, link)
continue
self._save_progress(page, i + 1, link)
if link in cache:
continue
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()
Path("progress.txt").unlink(missing_ok=True)
def main() -> None: def main() -> None:
if len(argv) != 2: if len(argv) != 3:
raise ValueError(f"{argv[0]} <sous-url>") raise ValueError(f"{argv[0]} <filename> <sous-url>")
filename = argv[1]
suburl = argv[2]
scraper: Scraper = Scraper() scraper: Scraper = Scraper()
scraper.getvins(argv[1], "donnee.csv") scraper.getvins(suburl, filename)
if __name__ == "__main__": if __name__ == "__main__":

View File

View File

@@ -319,7 +319,7 @@ def test_informations(scraper: Scraper):
def test_search(scraper: Scraper): def test_search(scraper: Scraper):
m = mock_open() m = mock_open()
with patch("builtins.open", m): with patch("builtins.open", m):
scraper.getvins("wine.html", "fake_file.csv") scraper.getvins("wine.html", "fake_file.csv", True)
assert m().write.called assert m().write.called
all_writes = "".join(call.args[0] for call in m().write.call_args_list) all_writes = "".join(call.args[0] for call in m().write.call_args_list)