Seguint P-2025-1, quan es prepara l’Assemblea General Ordinària, i de tant en tant, convé saber l’estat d’altes a l’associació.
Per evitar problemes de canvis de metodologia o errors històrics en les dades ho fem de la següent manera:
import os.path
from dolibarr import Dolibarr
import pprint
import csv
from decimal import Decimal
from typing import (
List,
Any,
Union,
Optional,
Dict,
Tuple,
Type,
TypeVar,
Callable,
cast,
)
import psycopg2
import attr
from decimal import Decimal
import datetime
import os
import os.path
import json
from io import StringIO
pp = pprint.PrettyPrinter()
def dateStr(x):
return x.strftime("%Y-%m-%d")
@attr.s
class MembershipPeriod:
begin: datetime.date = attr.ib()
end: datetime.date = attr.ib()
ref: str = attr.ib()
_wiggle_room = datetime.timedelta(days=2)
def matches_date(self, ref_date: datetime.date) -> bool:
return (
self.begin - MembershipPeriod._wiggle_room <= ref_date
and ref_date <= self.end + MembershipPeriod._wiggle_room
)
@attr.s
class Member:
ref: str = attr.ib()
active: bool = attr.ib()
firstname: str = attr.ib()
lastname: str = attr.ib()
is_person: bool = attr.ib()
entity_name: str = attr.ib()
entity_ref: str = attr.ib()
membership_periods: List[MembershipPeriod] = attr.ib(factory=list)
def status_on_date(self, ref_date: datetime.date) -> str:
for p in self.membership_periods:
if p.begin is None:
print(repr(self))
if p.matches_date(ref_date):
return p.ref
return ""
def get_membership_periods(self, dbconn):
with dbconn.cursor() as cursor:
cursor.execute(
"select f.datef, f.fk_soc, fd.qty, fd.date_start, fd.date_end, s.ref "
"from llx_facture as f "
"left join llx_facturedet as fd on f.rowid=fd.fk_facture "
"left join llx_product as s on fd.fk_product=s.rowid "
"where f.fk_soc = %s and s.ref like 'Q%%'"
"order by f.datef",
(self.entity_ref,),
)
o = []
first_date, last_date, cur_s = None, None, None
for f_date, _, s_qty, s_date_start, s_date_end, s in cursor.fetchall():
if first_date is None:
first_date = f_date
last_date = f_date
cur_s = s
period_begin, period_end = self._simplify_dates(
f_date, s_qty, s_date_start, s_date_end
)
if period_begin - last_date < datetime.timedelta(days=7) and cur_s == s:
last_date = period_end
else:
o.append(
MembershipPeriod(begin=first_date, end=last_date, ref=cur_s)
)
first_date, last_date, cur_s = period_begin, period_end, s
if first_date and last_date and cur_s:
o.append(MembershipPeriod(begin=first_date, end=last_date, ref=cur_s))
return o
def _simplify_dates(self, f_date, s_qty, s_date_start, s_date_end):
if s_date_start and s_date_end:
return (s_date_start.date(), s_date_end.date())
int_qty = int(s_qty)
rest_qty = s_qty - int_qty
end_year = f_date.year + (f_date.month + int_qty) // 12
end_month = (f_date.month + int_qty) % 12
if end_month == 0:
end_month = 12
end_year += 1
end_day = max(1, int(rest_qty * 28))
return (f_date, datetime.date(year=end_year, month=end_month, day=end_day))
def get_members(dbconn) -> List[Member]:
with dbconn.cursor() as cursor:
cursor.execute(
"SELECT rowid,statut,firstname,lastname,morphy,societe,fk_soc FROM llx_adherent"
)
o: List[Member] = []
for (
rowid,
statut,
firstname,
lastname,
morphy,
societe,
fk_soc,
) in cursor.fetchall():
member = Member(
ref=rowid,
active=(statut in [1, "1", True]),
firstname=firstname,
lastname=lastname,
is_person=(morphy.strip() == "phy"),
entity_name=societe,
entity_ref=fk_soc,
)
member.membership_periods = member.get_membership_periods(dbconn)
o.append(member)
return o
def summary_at_date(members: List[Member], ref_date: datetime.date):
filtered = filter(lambda m: m.status_on_date(ref_date), members)
o = {"S": 0, "E": 0, "S_Q0": 0, "S_Q1": 0, "E_Q0": 0, "E_Q1": 0, "Tot": 0}
for m in filtered:
prefix = "S" if m.is_person else "E"
bucket = f"{prefix}_{m.status_on_date(ref_date)}"
if bucket not in o:
o[bucket] = 0
o[bucket] += 1
o[prefix] += 1
o["Tot"] += 1
return o
def get_members_overview(
doliUrl: str,
doliApiToken: str,
pgHost: str,
pgPort: str,
pgDb: str,
pgUser: str,
pgPass: str,
) -> List[Any]:
with psycopg2.connect(
host=pgHost, port=pgPort, database=pgDb, user=pgUser, password=pgPass
) as dbconn:
dbconn.set_client_encoding("UNICODE")
return get_members(dbconn)
from typing import (
List,
Any,
Union,
Optional,
Dict,
Tuple,
Type,
TypeVar,
Callable,
cast,
)
import psycopg2
import attr
from decimal import Decimal
import datetime
import os
import os.path
import json
from io import StringIO
@attr.s
class MembershipPeriod:
begin: datetime.date = attr.ib()
end: datetime.date = attr.ib()
ref: str = attr.ib()
_wiggle_room = datetime.timedelta(days=2)
def matches_date(self, ref_date: datetime.date) -> bool:
return (
self.begin - MembershipPeriod._wiggle_room <= ref_date
and ref_date <= self.end + MembershipPeriod._wiggle_room
)
@attr.s
class Member:
ref: str = attr.ib()
active: bool = attr.ib()
firstname: str = attr.ib()
lastname: str = attr.ib()
is_person: bool = attr.ib()
entity_name: str = attr.ib()
entity_ref: str = attr.ib()
membership_periods: List[MembershipPeriod] = attr.ib(factory=list)
def status_on_date(self, ref_date: datetime.date) -> str:
for p in self.membership_periods:
if p.begin is None:
print(repr(self))
if p.matches_date(ref_date):
return p.ref
return ""
def get_membership_periods(self, dbconn):
with dbconn.cursor() as cursor:
cursor.execute(
"select f.datef, f.fk_soc, fd.qty, fd.date_start, fd.date_end, s.ref "
"from llx_facture as f "
"left join llx_facturedet as fd on f.rowid=fd.fk_facture "
"left join llx_product as s on fd.fk_product=s.rowid "
"where f.fk_soc = %s and s.ref like 'Q%%'"
"order by f.datef",
(self.entity_ref,),
)
o = []
first_date, last_date, cur_s = None, None, None
for f_date, _, s_qty, s_date_start, s_date_end, s in cursor.fetchall():
if first_date is None:
first_date = f_date
last_date = f_date
cur_s = s
period_begin, period_end = self._simplify_dates(
f_date, s_qty, s_date_start, s_date_end
)
if period_begin - last_date < datetime.timedelta(days=7) and cur_s == s:
last_date = period_end
else:
o.append(
MembershipPeriod(begin=first_date, end=last_date, ref=cur_s)
)
first_date, last_date, cur_s = period_begin, period_end, s
if first_date and last_date and cur_s:
o.append(MembershipPeriod(begin=first_date, end=last_date, ref=cur_s))
return o
def _simplify_dates(self, f_date, s_qty, s_date_start, s_date_end):
if s_date_start and s_date_end:
return (s_date_start.date(), s_date_end.date())
int_qty = int(s_qty)
rest_qty = s_qty - int_qty
end_year = f_date.year + (f_date.month + int_qty) // 12
end_month = (f_date.month + int_qty) % 12
if end_month == 0:
end_month = 12
end_year += 1
end_day = max(1, int(rest_qty * 28))
return (f_date, datetime.date(year=end_year, month=end_month, day=end_day))
def get_members(dbconn) -> List[Member]:
with dbconn.cursor() as cursor:
cursor.execute(
"SELECT rowid,statut,firstname,lastname,morphy,societe,fk_soc FROM llx_adherent"
)
o: List[Member] = []
for (
rowid,
statut,
firstname,
lastname,
morphy,
societe,
fk_soc,
) in cursor.fetchall():
member = Member(
ref=rowid,
active=(statut in [1, "1", True]),
firstname=firstname,
lastname=lastname,
is_person=(morphy.strip() == "phy"),
entity_name=societe,
entity_ref=fk_soc,
)
member.membership_periods = member.get_membership_periods(dbconn)
o.append(member)
return o
def summary_at_date(members: List[Member], ref_date: datetime.date):
filtered = filter(lambda m: m.status_on_date(ref_date), members)
o = {"S": 0, "E": 0, "S_Q0": 0, "S_Q1": 0, "E_Q0": 0, "E_Q1": 0, "Tot": 0}
for m in filtered:
prefix = "S" if m.is_person else "E"
bucket = f"{prefix}_{m.status_on_date(ref_date)}"
if bucket not in o:
o[bucket] = 0
o[bucket] += 1
o[prefix] += 1
o["Tot"] += 1
return o
def get_members_overview(
doliUrl: str,
doliApiToken: str,
pgHost: str,
pgPort: str,
pgDb: str,
pgUser: str,
pgPass: str,
) -> List[Any]:
with psycopg2.connect(
host=pgHost, port=pgPort, database=pgDb, user=pgUser, password=pgPass
) as dbconn:
dbconn.set_client_encoding("UNICODE")
return get_members(dbconn)
from typing import (
List,
Any,
Union,
Optional,
Dict,
Tuple,
Type,
TypeVar,
Callable,
cast,
)
import psycopg2
import attr
from decimal import Decimal
import datetime
import os
import os.path
import json
from io import StringIO
@attr.s
class MembershipPeriod:
begin: datetime.date = attr.ib()
end: datetime.date = attr.ib()
ref: str = attr.ib()
_wiggle_room = datetime.timedelta(days=2)
def matches_date(self, ref_date: datetime.date) -> bool:
return (
self.begin - MembershipPeriod._wiggle_room <= ref_date
and ref_date <= self.end + MembershipPeriod._wiggle_room
)
@attr.s
class Member:
ref: str = attr.ib()
active: bool = attr.ib()
firstname: str = attr.ib()
lastname: str = attr.ib()
is_person: bool = attr.ib()
entity_name: str = attr.ib()
entity_ref: str = attr.ib()
membership_periods: List[MembershipPeriod] = attr.ib(factory=list)
def status_on_date(self, ref_date: datetime.date) -> str:
for p in self.membership_periods:
if p.begin is None:
print(repr(self))
if p.matches_date(ref_date):
return p.ref
return ""
def get_membership_periods(self, dbconn):
with dbconn.cursor() as cursor:
cursor.execute(
"select f.datef, f.fk_soc, fd.qty, fd.date_start, fd.date_end, s.ref "
"from llx_facture as f "
"left join llx_facturedet as fd on f.rowid=fd.fk_facture "
"left join llx_product as s on fd.fk_product=s.rowid "
"where f.fk_soc = %s and s.ref like 'Q%%'"
"order by f.datef",
(self.entity_ref,),
)
o = []
first_date, last_date, cur_s = None, None, None
for f_date, _, s_qty, s_date_start, s_date_end, s in cursor.fetchall():
if first_date is None:
first_date = f_date
last_date = f_date
cur_s = s
period_begin, period_end = self._simplify_dates(
f_date, s_qty, s_date_start, s_date_end
)
if period_begin - last_date < datetime.timedelta(days=7) and cur_s == s:
last_date = period_end
else:
o.append(
MembershipPeriod(begin=first_date, end=last_date, ref=cur_s)
)
first_date, last_date, cur_s = period_begin, period_end, s
if first_date and last_date and cur_s:
o.append(MembershipPeriod(begin=first_date, end=last_date, ref=cur_s))
return o
def _simplify_dates(self, f_date, s_qty, s_date_start, s_date_end):
if s_date_start and s_date_end:
return (s_date_start.date(), s_date_end.date())
int_qty = int(s_qty)
rest_qty = s_qty - int_qty
end_year = f_date.year + (f_date.month + int_qty) // 12
end_month = (f_date.month + int_qty) % 12
if end_month == 0:
end_month = 12
end_year += 1
end_day = max(1, int(rest_qty * 28))
return (f_date, datetime.date(year=end_year, month=end_month, day=end_day))
def get_members(dbconn) -> List[Member]:
with dbconn.cursor() as cursor:
cursor.execute(
"SELECT rowid,statut,firstname,lastname,morphy,societe,fk_soc FROM llx_adherent"
)
o: List[Member] = []
for (
rowid,
statut,
firstname,
lastname,
morphy,
societe,
fk_soc,
) in cursor.fetchall():
member = Member(
ref=rowid,
active=(statut in [1, "1", True]),
firstname=firstname,
lastname=lastname,
is_person=(morphy.strip() == "phy"),
entity_name=societe,
entity_ref=fk_soc,
)
member.membership_periods = member.get_membership_periods(dbconn)
o.append(member)
return o
def summary_at_date(members: List[Member], ref_date: datetime.date):
filtered = filter(lambda m: m.status_on_date(ref_date), members)
o = {
"Persones": 0,
"Entitats": 0,
"P_Q0": 0,
"P_Q1": 0,
"E_Q0": 0,
"E_Q1": 0,
"Total": 0,
}
for m in filtered:
prefix = "P" if m.is_person else "E"
bucket = f"{prefix}_{m.status_on_date(ref_date)}"
if bucket not in o:
o[bucket] = 0
o[bucket] += 1
o["Persones" if m.is_person else "Entitats"] += 1
o["Total"] += 1
return o
def get_members_overview(
doliUrl: str,
doliApiToken: str,
pgHost: str,
pgPort: str,
pgDb: str,
pgUser: str,
pgPass: str,
) -> List[Any]:
with psycopg2.connect(
host=pgHost, port=pgPort, database=pgDb, user=pgUser, password=pgPass
) as dbconn:
dbconn.set_client_encoding("UNICODE")
return get_members(dbconn)
def members_overview(**kwargs):
r = get_members_overview(**kwargs)
today = datetime.date.today()
start_date = datetime.date(year=2020, month=5, day=1) # Start of data is 2020
ref_dates = [
max(start_date, datetime.date(year=year, month=1, day=1))
for year in range(start_date.year, today.year + 1)
]
ref_dates.append(today)
data = []
for ref_date in ref_dates:
summary = summary_at_date(r, ref_date)
data.append(summary)
return list(zip(ref_dates, data))
def main():
import os
import sys
r = members_overview(
doliUrl=os.environ.get("DOLI_API_URL", "http://localhost:8080/api/index.php/"),
doliApiToken=os.environ["DOLI_API_TOKEN"],
pgHost=os.environ.get("DB_HOST", "localhost"),
pgPort=os.environ.get("DB_PORT", "5432"),
pgDb=os.environ.get("DB_NAME", "dolibarr"),
pgUser=os.environ.get("DB_USER", "postgres"),
pgPass=os.environ["DB_PASS"],
)
ks = list(sorted(set((k for d in r for k in d[1].keys()))))
if "--csv" in sys.argv:
print("#" + ",".join(["date"] + ks))
for d, data in r:
print(",".join([dateStr(d)] + [repr(data.get(k, "")) for k in ks]))
if "--md" in sys.argv:
print(
"""
- Metodologia: https://agora.exo.cat/t/p-2025-22-cens-dassociacions-a-lexo/400
- Q0 són quotes bàsiques (3€/mes)
- Q1 són quotes contributives (10€/mes)"""
)
print("| " + " | ".join(["Data"] + ks) + " |")
print("| " + " --- |" * (len(ks) + 1))
for d, data in r:
print(
"| "
+ " | ".join([dateStr(d)] + [repr(data.get(k, "")) for k in ks])
+ " |"
)
else:
pp.pprint(r)
return r
if __name__ == "__main__":
main()