Fix exports Apogée (pb liés passage à Python 3)

This commit is contained in:
Emmanuel Viennet 2022-01-07 17:58:34 +01:00
parent 4140f11b7b
commit d68f81b4ea
6 changed files with 39 additions and 26 deletions

View File

@ -108,13 +108,14 @@ def apo_compare_csv(A_file, B_file, autodetect=True):
def _load_apo_data(csvfile, autodetect=True):
"Read data from request variable and build ApoData"
data = csvfile.read()
data_b = csvfile.read()
if autodetect:
data, message = sco_apogee_csv.fix_data_encoding(data)
data_b, message = sco_apogee_csv.fix_data_encoding(data_b)
if message:
log("apo_compare_csv: %s" % message)
if not data:
if not data_b:
raise ScoValueError("apo_compare_csv: no data")
data = data_b.decode(sco_apogee_csv.APO_INPUT_ENCODING)
apo_data = sco_apogee_csv.ApoData(data, orig_filename=csvfile.filename)
return apo_data

View File

@ -173,8 +173,10 @@ def guess_data_encoding(text, threshold=0.6):
def fix_data_encoding(
text, default_source_encoding=APO_INPUT_ENCODING, dest_encoding=APO_INPUT_ENCODING
):
text: bytes,
default_source_encoding=APO_INPUT_ENCODING,
dest_encoding=APO_INPUT_ENCODING,
) -> bytes:
"""Try to ensure that text is using dest_encoding
returns converted text, and a message describing the conversion.
"""
@ -200,7 +202,7 @@ def fix_data_encoding(
class StringIOFileLineWrapper(object):
def __init__(self, data):
def __init__(self, data: str):
self.f = io.StringIO(data)
self.lineno = 0
@ -655,7 +657,7 @@ class ApoEtud(dict):
class ApoData(object):
def __init__(
self,
data,
data: str,
periode=None,
export_res_etape=True,
export_res_sem=True,
@ -693,7 +695,7 @@ class ApoData(object):
"<h3>Erreur lecture du fichier Apogée <tt>%s</tt></h3><p>" % filename
+ e.args[0]
+ "</p>"
)
) from e
self.etape_apogee = self.get_etape_apogee() # 'V1RT'
self.vdi_apogee = self.get_vdi_apogee() # '111'
self.etape = ApoEtapeVDI(etape=self.etape_apogee, vdi=self.vdi_apogee)
@ -760,7 +762,6 @@ class ApoData(object):
def read_csv(self, data: str):
if not data:
raise ScoFormatError("Fichier Apogée vide !")
f = StringIOFileLineWrapper(data) # pour traiter comme un fichier
# check that we are at the begining of Apogee CSV
line = f.readline().strip()
@ -768,7 +769,10 @@ class ApoData(object):
raise ScoFormatError("format incorrect: pas de XX-APO_TITRES-XX")
# 1-- En-tête: du début jusqu'à la balise XX-APO_VALEURS-XX
try:
idx = data.index("XX-APO_VALEURS-XX")
except ValueError as exc:
raise ScoFormatError("format incorrect: pas de XX-APO_VALEURS-XX") from exc
self.header = data[:idx]
# 2-- Titres:
@ -1178,7 +1182,7 @@ def nar_etuds_table(apo_data, NAR_Etuds):
def export_csv_to_apogee(
apo_csv_data,
apo_csv_data: str,
periode=None,
dest_zip=None,
export_res_etape=True,

View File

@ -43,7 +43,7 @@
apo_csv_get()
API:
apo_csv_store( annee_scolaire, sem_id)
# apo_csv_store(csv_data, annee_scolaire, sem_id)
store maq file (archive)
apo_csv_get(etape_apo, annee_scolaire, sem_id, vdi_apo=None)
@ -101,7 +101,7 @@ ApoCSVArchive = ApoCSVArchiver()
def apo_csv_store(csv_data: str, annee_scolaire, sem_id):
"""
csv_data: maquette content (string)
csv_data: maquette content (str))
annee_scolaire: int (2016)
sem_id: 0 (année ?), 1 (premier semestre de l'année) ou 2 (deuxième semestre)
:return: etape_apo du fichier CSV stocké
@ -378,7 +378,7 @@ e.associate_sco( apo_data)
print apo_csv_list_stored_archives()
apo_csv_store(csv_data, annee_scolaire, sem_id)
# apo_csv_store(csv_data, annee_scolaire, sem_id)

View File

@ -48,7 +48,7 @@ from app.scodoc import sco_preferences
from app.scodoc import sco_semset
from app.scodoc import sco_etud
from app.scodoc.gen_tables import GenTable
from app.scodoc.sco_apogee_csv import APO_PORTAL_ENCODING, APO_INPUT_ENCODING
from app.scodoc.sco_apogee_csv import APO_INPUT_ENCODING, APO_OUTPUT_ENCODING
from app.scodoc.sco_exceptions import ScoValueError
@ -585,7 +585,7 @@ def _view_etuds_page(semset_id, title="", etuds=[], keys=(), format="html"):
return "\n".join(H) + html_sco_header.sco_footer()
def view_apo_csv_store(semset_id="", csvfile=None, data="", autodetect=False):
def view_apo_csv_store(semset_id="", csvfile=None, data: bytes = "", autodetect=False):
"""Store CSV data
Le semset identifie l'annee scolaire et le semestre
Si csvfile, lit depuis FILE, sinon utilise data
@ -593,9 +593,8 @@ def view_apo_csv_store(semset_id="", csvfile=None, data="", autodetect=False):
if not semset_id:
raise ValueError("invalid null semset_id")
semset = sco_semset.SemSet(semset_id=semset_id)
if csvfile:
data = csvfile.read()
data = csvfile.read() # bytes
if autodetect:
# check encoding (although documentation states that users SHOULD upload LATIN1)
data, message = sco_apogee_csv.fix_data_encoding(data)
@ -605,19 +604,26 @@ def view_apo_csv_store(semset_id="", csvfile=None, data="", autodetect=False):
log("view_apo_csv_store: autodetection of encoding disabled by user")
if not data:
raise ScoValueError("view_apo_csv_store: no data")
# data est du bytes, encodé en APO_INPUT_ENCODING
data_str = data.decode(APO_INPUT_ENCODING)
# check si etape maquette appartient bien au semset
apo_data = sco_apogee_csv.ApoData(
data, periode=semset["sem_id"]
data_str, periode=semset["sem_id"]
) # parse le fichier -> exceptions
if apo_data.etape not in semset["etapes"]:
raise ScoValueError(
"Le code étape de ce fichier ne correspond pas à ceux de cet ensemble"
)
sco_etape_apogee.apo_csv_store(data, semset["annee_scolaire"], semset["sem_id"])
sco_etape_apogee.apo_csv_store(data_str, semset["annee_scolaire"], semset["sem_id"])
return flask.redirect("apo_semset_maq_status?semset_id=" + semset_id)
return flask.redirect(
url_for(
"notes.apo_semset_maq_status",
scodoc_dept=g.scodoc_dept,
semset_id=semset_id,
)
)
def view_apo_csv_download_and_store(etape_apo="", semset_id=""):
@ -629,9 +635,9 @@ def view_apo_csv_download_and_store(etape_apo="", semset_id=""):
data = sco_portal_apogee.get_maquette_apogee(
etape=etape_apo, annee_scolaire=semset["annee_scolaire"]
)
# here, data is utf8
# here, data is str
# but we store and generate latin1 files, to ease further import in Apogée
data = data.decode(APO_PORTAL_ENCODING).encode(APO_INPUT_ENCODING) # XXX #py3
data = data.encode(APO_OUTPUT_ENCODING)
return view_apo_csv_store(semset_id, data=data, autodetect=False)
@ -669,7 +675,7 @@ def view_apo_csv(etape_apo="", semset_id="", format="html"):
sem_id = semset["sem_id"]
csv_data = sco_etape_apogee.apo_csv_get(etape_apo, annee_scolaire, sem_id)
if format == "raw":
scu.send_file(csv_data, etape_apo, suffix=".txt", mime=scu.CSV_MIMETYPE)
return scu.send_file(csv_data, etape_apo, suffix=".txt", mime=scu.CSV_MIMETYPE)
apo_data = sco_apogee_csv.ApoData(csv_data, periode=semset["sem_id"])

View File

@ -544,7 +544,7 @@ def check_paiement_etuds(etuds):
etud["paiementinscription_str"] = "(pb cnx Apogée)"
def get_maquette_apogee(etape="", annee_scolaire=""):
def get_maquette_apogee(etape="", annee_scolaire="") -> str:
"""Maquette CSV Apogee pour une étape et une annee scolaire"""
maquette_url = get_maquette_url()
if not maquette_url:

View File

@ -395,6 +395,8 @@ def do_semset_add_sem(semset_id, formsemestre_id):
"""Add a sem to a semset"""
if not semset_id:
raise ScoValueError("empty semset_id")
if formsemestre_id == "":
raise ScoValueError("pas de semestre choisi !")
s = SemSet(semset_id=semset_id)
# check for valid formsemestre_id
_ = sco_formsemestre.get_formsemestre(formsemestre_id) # raise exc