1729921193

Build Type

github

Committed by GitHub

Commit Message

Merge 214c8c6e0 into 83b7b2b5a

Pull Request Pull Request #16: Upload atc cases

Run Details

416 of 1161 branches covered (35.83%)

Branch coverage included in aggregate %.

42 of 92 new or added lines in 4 files covered. (45.65%)

3 existing lines in 1 file now uncovered.

1244 of 3046 relevant lines covered (40.84%)

0.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

36.26

/tomatic/api.py

# -*- coding: utf-8 -*-

from fastapi import (
    FastAPI,
    Request,
    Form,
    WebSocket,
    WebSocketDisconnect,
    File,
    UploadFile,
)
from fastapi.responses import (
    FileResponse,
    Response,
)
from . import __version__ as version
import asyncio
import re
from datetime import datetime, timedelta, timezone
import urllib.parse
import decorator
import erppeek
from pathlib import Path
from yamlns import namespace as ns
from consolemsg import error, step, warn, u

from .callinfo import CallInfo
from .callregistry import CallRegistry
from . import schedulestorage
from .pbx import pbxqueue as pbx
from . import persons
from .backchannel import BackChannel

try:
    import dbconfig
except ImportError:
    dbconfig = None


packagedir = Path(__file__).parent
distpath = packagedir/'dist'
schedules = schedulestorage.Storage.default()

def fillConfigurationInfo():
    return ns.load('config.yaml')
CONFIG = fillConfigurationInfo()

from contextlib import contextmanager
@contextmanager
def erp():
    if not hasattr(erp,'clients'):
        erp.clients = []
        erp.available = []
    if not erp.available:
        newclient = erppeek.Client(**dbconfig.erppeek)
        erp.clients.append(newclient)
        erp.available.append(newclient)
    client = erp.available.pop()
    try:
        yield client
    finally:
        erp.available.append(client)

def now():
    return datetime.now()

def anow():
    return datetime(2020,11,9,10,23,23)

def thisweek():
    return format(now().date() - timedelta(days=now().weekday()))

from .planner_api import api as Planner
from fastapi.websockets import WebSocket

app = FastAPI()
app.include_router(Planner, prefix='/api/planner')


class ApiError(Exception): pass

def yamlfy(status=200, data=[], **kwd):
    output = ns(data, **kwd)
    return Response(output.dump(),
        status,
        media_type = 'application/x-yaml',
    )

@decorator.decorator
async def ayamlerrors(f,*args,**kwd):
    try:
        return await f(*args,**kwd)
    except ApiError as e:
        error("ApiError: {}", e)
        return yamlfy(
            error=format(e),
            status=400,
            )
    except Exception as e:
        error("UnexpectedError: {}", e)
        import traceback
        error(''.join(traceback.format_exc()))
        return yamlfy(
            error=format(e),
            status=500,
            )


@decorator.decorator
def yamlerrors(f,*args,**kwd):
    try:
        return f(*args,**kwd)
    except ApiError as e:
        error("ApiError: {}", e)
        return yamlfy(
            error=format(e),
            status=400,
            )
    except Exception as e:
        error("UnexpectedError: {}", e)
        import traceback
        error(''.join(traceback.format_exc()))
        return yamlfy(
            error=format(e),
            status=500,
            )

backchannel = BackChannel()

@app.websocket('/backchannel')
async def websocketSession(websocket: WebSocket):
    await websocket.accept()
    session_id = str(websocket.client)
    def sender(message):
        step("WS {} send: {}", session_id, message)
        return websocket.send_text(message)

    backchannel.addSession(session_id, websocket, sender)
    try:
        while True:
            data = await websocket.receive_text()
            backchannel.receiveMessage(session_id, data)
    except WebSocketDisconnect:
        backchannel.onDisconnect(session_id)


@app.get('/')
@app.get('/{file}')
def tomatic(file=None):
    return FileResponse(distpath / (file or 'index.html'))

@app.get('/api/version')
@yamlerrors
def apiVersion():
    return yamlfy(
        version = version,
    )

@app.get('/api/graella/list')
@yamlerrors
def listGraelles():
    return yamlfy(weeks=schedules.list())

@app.get('/api/graella/retireold')
@yamlerrors
def retireOldTimeTable():
    today = datetime.today()
    twoMondaysAgo = str((today - timedelta(days=today.weekday()+7*2)).date())
    step("Retiring timetables older than {}", twoMondaysAgo)
    try:
        schedules.load(twoMondaysAgo)
    except:
        raise ApiError("Graella {} no trobada, potser ja ha estat retirada".format(twoMondaysAgo))
    else:
        schedules.retireOld(twoMondaysAgo)
    return yamlfy(result='ok')


@app.get('/api/graella-{week}.yaml')
@app.get('/api/graella/{week}')
@yamlerrors
def graellaYaml(week):
    schedule = schedules.load(week)

    return yamlfy(**schedule)

@app.patch('/api/graella/{week}/{day}/{houri}/{turni}/{name}')
@ayamlerrors
async def editSlot(week, day, houri: int, turni: int, name, request: Request):
    # TODO: This should be some kind of auth
    user = (await request.body()).decode('utf8').split('"')[1]
    graella = schedules.load(week)
    # TODO: Ensure day, houri, turni and name are in graella
    oldName = graella.timetable[day][int(houri)][int(turni)]
    graella.timetable[day][int(houri)][int(turni)] = name
    graella.overload = graella.get('overload', ns())
    graella.overload[oldName] = graella.overload.get(oldName, 0) -1
    graella.overload[name] = graella.overload.get(name, 0) +1
    logmsg = (
        "{}: {} ha canviat {} {}-{} {} de {} a {}".format(
        datetime.now(),
        user, # TODO: ERP user
        day,
        graella.hours[int(houri)],
        graella.hours[int(houri)+1],
        graella.turns[int(turni)],
        oldName,
        name
        ))
    step(logmsg)
    graella.setdefault('log',[]).append(logmsg)
    schedules.save(graella)
    schedulestorage.publishStatic(graella)
    return graellaYaml(week)


def cachedQueueStatus(force=False):
    now = datetime.now()
    if not force and hasattr(cachedQueueStatus, 'value') and cachedQueueStatus.timestamp > now:
        return cachedQueueStatus.value
    cachedQueueStatus.timestamp = now + timedelta(seconds=5)
    cachedQueueStatus.value = pbx().queue()
    return cachedQueueStatus.value

@app.post('/api/graella')
@yamlerrors
def uploadGraella(yaml: UploadFile = File(...), week=None):
    step("uploading {}".format(yaml.filename))
    graella = ns.load(yaml.file)
    logmsg = (
        "{}: {} ha pujat {} ".format(
        datetime.now(),
        "nobody", # TODO: ERP user
        graella.week,
        ))
    graella.setdefault('log',[]).append(logmsg)
    schedules.save(graella)
    schedulestorage.publishStatic(graella)
    return yamlfy(result='ok')

@app.get('/api/queue')
@yamlerrors
def get_queue():
    return yamlfy(
        currentQueue = cachedQueueStatus()
    )

@app.get('/api/queue/add/{person}')
@yamlerrors
def add_line(person):
    p = pbx()
    p.add(person)
    return yamlfy(
        currentQueue = cachedQueueStatus(force=True)
    )

@app.get('/api/queue/pause/{person}')
@yamlerrors
def pause_line(person):
    p = pbx()
    p.pause(person)
    return yamlfy(
        currentQueue = cachedQueueStatus(force=True)
    )

@app.get('/api/queue/resume/{person}')
@yamlerrors
def resume_line(person):
    p = pbx()
    p.resume(person)
    return yamlfy(
        currentQueue = cachedQueueStatus(force=True)
    )

@app.get('/api/persons/extension/{extension}')
@yamlerrors
def personInfoFromExtension(extension):
    allpersons=persons.persons()
    names = [name for name,ext in allpersons.extensions.items() if ext == extension]
    if not names:
        return 'nobody@somenergia.coop'
    name = names[0]
    email = allpersons.emails[name]
    return email

@app.get('/api/persons')
@yamlerrors
def personInfo():
    result=persons.persons()
    return yamlfy(persons=result)

@app.post('/api/person/{person}')
@ayamlerrors
async def setPersonInfo(person, request: Request):
    data = ns.loads(await request.body())
    persons.update(person, data)
    return yamlfy(persons=persons.persons())

@app.get('/api/busy/{person}')
@yamlerrors
def busy(person):
    from . import busy
    return yamlfy(**busy.busy(person))

@app.post('/api/busy/{person}')
@ayamlerrors
async def busy_post(person, request: Request):
    from . import busy
    data = ns.loads(await request.body())
    return yamlfy(**busy.update_busy(person, data))

@app.get('/api/busy/download/weekly')
@yamlerrors
def downloadWeeklyBusy():
    response = FileResponse(
        path='indisponibilitats.conf',
        #as_attachment=True,
        media_type='text/plain',
    )
    return response

@app.get('/api/busy/download/oneshot')
@yamlerrors
def downloadOneShotBusy():
    return FileResponse(
        'oneshot.conf',
        #as_attachment=True,
        media_type='text/plain',
    )

@app.get('/api/shifts/download/credit/{week}')
@yamlerrors
def downloadWeekShiftCredit(week):
    try:
        credit = schedules.credit(week)
    except schedulestorage.StorageError as e:
        raise ApiError(e)
    return yamlfy(**credit)

@app.get('/api/shifts/download/shiftload/{week}')
def downloadShiftLoad(week):
    loadfile = Path('carrega-{}.csv'.format(week))

    return FileResponse(
        str(loadfile),
        #as_attachment=True,
        media_type='text/csv',
    )

@app.get('/api/shifts/download/overload/{week}')
def downloadOverload(week):
    loadfile = Path('overload-{}.yaml'.format(week))

    return FileResponse(
        str(loadfile),
        #as_attachment=True,
        media_type = 'application/x-yaml',
    )

def yamlinfoerror(code, message, *args, **kwds):
    error(message, *args, **kwds)
    return yamlfy(info=ns(
        info='',
        message=code,
    ))

@app.get('/api/info/{field}/{value}')
def getInfoPersonBy(field, value):
    decoded_field = urllib.parse.unquote(value)
    data = None
    with erp() as O:
        callinfo = CallInfo(O)
        try:
            data = callinfo.getByField(field, decoded_field, shallow=True)
        except ValueError:
            return yamlinfoerror('error_getBy'+field.title(),
                "Getting information searching {}='{}'.", field, value)
    if data is None:
        return 404
    message = 'ok'
    if not data.partners:
        message = 'no_info'
    elif data.partners == "Masses resultats":
        message = 'response_too_long'

    result = ns(
        info=data,
        message=message,
    )
    return yamlfy(info=result)

@app.post('/api/info/contractdetails')
async def getContractDetails(request: Request):
    params = ns.loads(await request.body())
    with erp() as O:
        info = CallInfo(O)
        data = info.contractDetails(params.contracts)

    result = ns(
        info=data,
        message='ok',
    )
    return yamlfy(info=result)


@app.get('/api/info/ringring')
async def callingPhone(phone: str, extension: str):
    return await notifyIncommingCall(phone, extension)

@app.post('/api/info/ringring')
async def callingPhonePost(phone: str = Form(...), ext: str = Form(...)):
    return await notifyIncommingCall(phone, ext)

async def notifyIncommingCall(phone: str, extension: str):
    time = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
    user = persons.byExtension(extension)
    phone = phone.replace('+','').replace('-','').replace(' ','')
    phone = re.sub(r'^[+]?0?0?34','', phone)
    CallRegistry().updateCall(extension, fields=ns(
        data = time,
        telefon = phone,
        motius = "",
        partner = "",
        contracte = "",
    ))
    notifications = backchannel.notifyIncommingCall(user, phone, time)
    if not notifications:
        warn(
            f"No sesion on extension {extension} "
            f"to notify incomming call from {phone} at {time}")
    await asyncio.gather(*notifications)
    return yamlfy(result='ok')


@app.get('/api/personlog/{extension}')
def getCallLog(extension):
    calls = CallRegistry().callsByExtension(extension)
    return yamlfy(
        info=ns(
            info=calls,
            message='ok',
        )
    )

@app.post('/api/call/annotate')
async def callAnnotate(request: Request):
    annotation = ns.loads(await request.body())
    CallRegistry().annotateCall(annotation)
    user = annotation.get('user', None)
    if user:
        await asyncio.gather(
            *backchannel.notifyCallLogChanged(user)
        )
    return yamlfy(info=ns(
        message="ok"
    ))




@app.get('/api/updateClaims')
def updateClaimTypes():
    with erp() as O:
        CallRegistry().importClaimTypes(O)
    return yamlfy(info=ns(message='ok'))


@app.get('/api/getClaims')
def getClaimTypes():
    message = 'ok'
    claims = CallRegistry().claimTypes()
    if not claims:
        error("File of claims does not exist")
        return yamlfy(info=ns(
            message="error",
            claims=[],
            dict={},
        ))
    claims_dict = CallRegistry().claimKeywords()
    if not claims_dict:
        error("File of claims dict does not exist")
        return yamlfy(info=ns(
            message="error",
            claims=[],
            dict={},
        ))

    result = ns(
        message=message,
        claims=claims,
        dict=claims_dict,
    )
    return yamlfy(info=result)


@app.get('/api/updateCrmCategories')
def updateCrmCategories():
    with erp() as O:
        CallRegistry().importCrmCategories(O)
    return yamlfy(info=ns(message='ok'))


@app.get('/api/getInfos')
def getInfos():
    infos = CallRegistry().infoRequestTypes()
    if not infos:
        return yamlinfoerror("error",
            "Unable to info request types")
    return yamlfy(info = ns(
        message='ok',
        infos=infos,
    ))




# vim: ts=4 sw=4 et

1	# -- coding: utf-8 --
2
3	from fastapi import (	2✔
4	FastAPI,
5	Request,
6	Form,
7	WebSocket,
8	WebSocketDisconnect,
9	File,
10	UploadFile,
11	)
12	from fastapi.responses import (	2✔
13	FileResponse,
14	Response,
15	)
16	from . import __version__ as version	2✔
17	import asyncio	2✔
18	import re	2✔
19	from datetime import datetime, timedelta, timezone	2✔
20	import urllib.parse	2✔
21	import decorator	2✔
22	import erppeek	2✔
23	from pathlib import Path	2✔
24	from yamlns import namespace as ns	2✔
25	from consolemsg import error, step, warn, u	2✔
26
27	from .callinfo import CallInfo	2✔
28	from .callregistry import CallRegistry	2✔
29	from . import schedulestorage	2✔
30	from .pbx import pbxqueue as pbx	2✔
31	from . import persons	2✔
32	from .backchannel import BackChannel	2✔
33
34	try:	2✔
35	import dbconfig	2✔
36	except ImportError:	2✔
37	dbconfig = None	2✔
38
39
40	packagedir = Path(__file__).parent	2✔
41	distpath = packagedir/'dist'	2✔
42	schedules = schedulestorage.Storage.default()	2✔
43
44	def fillConfigurationInfo():	2✔
45	return ns.load('config.yaml')	2✔
46	CONFIG = fillConfigurationInfo()	2✔
47
48	from contextlib import contextmanager	2✔
49	@contextmanager	2✔
50	def erp():	2✔
51	if not hasattr(erp,'clients'):	×
52	erp.clients = []	×
53	erp.available = []	×
54	if not erp.available:	×
55	newclient = erppeek.Client(**dbconfig.erppeek)	×
56	erp.clients.append(newclient)	×
57	erp.available.append(newclient)	×
58	client = erp.available.pop()	×
59	try:	×
60	yield client	×
61	finally:
62	erp.available.append(client)	×
63
64	def now():	2✔
65	return datetime.now()	×
66
67	def anow():	2✔
68	return datetime(2020,11,9,10,23,23)	×
69
70	def thisweek():	2✔
71	return format(now().date() - timedelta(days=now().weekday()))	×
72
73	from .planner_api import api as Planner	2✔
74	from fastapi.websockets import WebSocket	2✔
75
76	app = FastAPI()	2✔
77	app.include_router(Planner, prefix='/api/planner')	2✔
78
79
80	class ApiError(Exception): pass	2✔
81
82	def yamlfy(status=200, data=[], **kwd):	2✔
83	output = ns(data, **kwd)	×
84	return Response(output.dump(),	×
85	status,
86	media_type = 'application/x-yaml',
87	)
88
89	@decorator.decorator	2✔
90	async def ayamlerrors(f,args,*kwd):	2✔
91	try:	×
92	return await f(args,*kwd)	×
93	except ApiError as e:	×
94	error("ApiError: {}", e)	×
95	return yamlfy(	×
96	error=format(e),
97	status=400,
98	)
99	except Exception as e:	×
100	error("UnexpectedError: {}", e)	×
101	import traceback	×
102	error(''.join(traceback.format_exc()))	×
103	return yamlfy(	×
104	error=format(e),
105	status=500,
106	)
107
108
109	@decorator.decorator	2✔
110	def yamlerrors(f,args,*kwd):	2✔
111	try:	×
112	return f(args,*kwd)	×
113	except ApiError as e:	×
114	error("ApiError: {}", e)	×
115	return yamlfy(	×
116	error=format(e),
117	status=400,
118	)
119	except Exception as e:	×
120	error("UnexpectedError: {}", e)	×
121	import traceback	×
122	error(''.join(traceback.format_exc()))	×
123	return yamlfy(	×
124	error=format(e),
125	status=500,
126	)
127
128	backchannel = BackChannel()	2✔
129
130	@app.websocket('/backchannel')	2✔
131	async def websocketSession(websocket: WebSocket):	2✔
132	await websocket.accept()	×
133	session_id = str(websocket.client)	×
134	def sender(message):	×
135	step("WS {} send: {}", session_id, message)	×
136	return websocket.send_text(message)	×
137
138	backchannel.addSession(session_id, websocket, sender)	×
139	try:	×
140	while True:	×
141	data = await websocket.receive_text()	×
142	backchannel.receiveMessage(session_id, data)	×
143	except WebSocketDisconnect:	×
144	backchannel.onDisconnect(session_id)	×
145
146
147	@app.get('/')	2✔
148	@app.get('/{file}')	2✔
149	def tomatic(file=None):	2✔
150	return FileResponse(distpath / (file or 'index.html'))	×
151
152	@app.get('/api/version')	2✔
153	@yamlerrors	2✔
154	def apiVersion():	2✔
155	return yamlfy(	×
156	version = version,
157	)
158
159	@app.get('/api/graella/list')	2✔
160	@yamlerrors	2✔
161	def listGraelles():	2✔
162	return yamlfy(weeks=schedules.list())	×
163
164	@app.get('/api/graella/retireold')	2✔
165	@yamlerrors	2✔
166	def retireOldTimeTable():	2✔
167	today = datetime.today()	×
168	twoMondaysAgo = str((today - timedelta(days=today.weekday()+7*2)).date())	×
169	step("Retiring timetables older than {}", twoMondaysAgo)	×
170	try:	×
171	schedules.load(twoMondaysAgo)	×
172	except:	×
173	raise ApiError("Graella {} no trobada, potser ja ha estat retirada".format(twoMondaysAgo))	×
174	else:
175	schedules.retireOld(twoMondaysAgo)	×
176	return yamlfy(result='ok')	×
177
178
179	@app.get('/api/graella-{week}.yaml')	2✔
180	@app.get('/api/graella/{week}')	2✔
181	@yamlerrors	2✔
182	def graellaYaml(week):	2✔
183	schedule = schedules.load(week)	×
184
185	return yamlfy(**schedule)	×
186
187	@app.patch('/api/graella/{week}/{day}/{houri}/{turni}/{name}')	2✔
188	@ayamlerrors	2✔
189	async def editSlot(week, day, houri: int, turni: int, name, request: Request):	2✔
190	# TODO: This should be some kind of auth
191	user = (await request.body()).decode('utf8').split('"')[1]	×
192	graella = schedules.load(week)	×
193	# TODO: Ensure day, houri, turni and name are in graella
194	oldName = graella.timetable[day][int(houri)][int(turni)]	×
195	graella.timetable[day][int(houri)][int(turni)] = name	×
196	graella.overload = graella.get('overload', ns())	×
197	graella.overload[oldName] = graella.overload.get(oldName, 0) -1	×
198	graella.overload[name] = graella.overload.get(name, 0) +1	×
199	logmsg = (	×
200	"{}: {} ha canviat {} {}-{} {} de {} a {}".format(
201	datetime.now(),
202	user, # TODO: ERP user
203	day,
204	graella.hours[int(houri)],
205	graella.hours[int(houri)+1],
206	graella.turns[int(turni)],
207	oldName,
208	name
209	))
210	step(logmsg)	×
211	graella.setdefault('log',[]).append(logmsg)	×
212	schedules.save(graella)	×
213	schedulestorage.publishStatic(graella)	×
214	return graellaYaml(week)	×
215
216
217	def cachedQueueStatus(force=False):	2✔
218	now = datetime.now()	×
219	if not force and hasattr(cachedQueueStatus, 'value') and cachedQueueStatus.timestamp > now:	×
220	return cachedQueueStatus.value	×
221	cachedQueueStatus.timestamp = now + timedelta(seconds=5)	×
222	cachedQueueStatus.value = pbx().queue()	×
223	return cachedQueueStatus.value	×
224
225	@app.post('/api/graella')	2✔
226	@yamlerrors	2✔
227	def uploadGraella(yaml: UploadFile = File(...), week=None):	2✔
228	step("uploading {}".format(yaml.filename))	×
229	graella = ns.load(yaml.file)	×
230	logmsg = (	×
231	"{}: {} ha pujat {} ".format(
232	datetime.now(),
233	"nobody", # TODO: ERP user
234	graella.week,
235	))
236	graella.setdefault('log',[]).append(logmsg)	×
237	schedules.save(graella)	×
238	schedulestorage.publishStatic(graella)	×
239	return yamlfy(result='ok')	×
240
241	@app.get('/api/queue')	2✔
242	@yamlerrors	2✔
243	def get_queue():	2✔
244	return yamlfy(	×
245	currentQueue = cachedQueueStatus()
246	)
247
248	@app.get('/api/queue/add/{person}')	2✔
249	@yamlerrors	2✔
250	def add_line(person):	2✔
251	p = pbx()	×
252	p.add(person)	×
253	return yamlfy(	×
254	currentQueue = cachedQueueStatus(force=True)
255	)
256
257	@app.get('/api/queue/pause/{person}')	2✔
258	@yamlerrors	2✔
259	def pause_line(person):	2✔
260	p = pbx()	×
261	p.pause(person)	×
262	return yamlfy(	×
263	currentQueue = cachedQueueStatus(force=True)
264	)
265
266	@app.get('/api/queue/resume/{person}')	2✔
267	@yamlerrors	2✔
268	def resume_line(person):	2✔
269	p = pbx()	×
270	p.resume(person)	×
271	return yamlfy(	×
272	currentQueue = cachedQueueStatus(force=True)
273	)
274
275	@app.get('/api/persons/extension/{extension}')	2✔
276	@yamlerrors	2✔
277	def personInfoFromExtension(extension):	2✔
278	allpersons=persons.persons()	×
279	names = [name for name,ext in allpersons.extensions.items() if ext == extension]	×
280	if not names:	×
281	return 'nobody@somenergia.coop'	×
282	name = names[0]	×
283	email = allpersons.emails[name]	×
284	return email	×
285
286	@app.get('/api/persons')	2✔
287	@yamlerrors	2✔
288	def personInfo():	2✔
289	result=persons.persons()	×
290	return yamlfy(persons=result)	×
291
292	@app.post('/api/person/{person}')	2✔
293	@ayamlerrors	2✔
294	async def setPersonInfo(person, request: Request):	2✔
295	data = ns.loads(await request.body())	×
296	persons.update(person, data)	×
297	return yamlfy(persons=persons.persons())	×
298
299	@app.get('/api/busy/{person}')	2✔
300	@yamlerrors	2✔
301	def busy(person):	2✔
302	from . import busy	×
303	return yamlfy(**busy.busy(person))	×
304
305	@app.post('/api/busy/{person}')	2✔
306	@ayamlerrors	2✔
307	async def busy_post(person, request: Request):	2✔
308	from . import busy	×
309	data = ns.loads(await request.body())	×
310	return yamlfy(**busy.update_busy(person, data))	×
311
312	@app.get('/api/busy/download/weekly')	2✔
313	@yamlerrors	2✔
314	def downloadWeeklyBusy():	2✔
315	response = FileResponse(	×
316	path='indisponibilitats.conf',
317	#as_attachment=True,
318	media_type='text/plain',
319	)
320	return response	×
321
322	@app.get('/api/busy/download/oneshot')	2✔
323	@yamlerrors	2✔
324	def downloadOneShotBusy():	2✔
325	return FileResponse(	×
326	'oneshot.conf',
327	#as_attachment=True,
328	media_type='text/plain',
329	)
330
331	@app.get('/api/shifts/download/credit/{week}')	2✔
332	@yamlerrors	2✔
333	def downloadWeekShiftCredit(week):	2✔
334	try:	×
335	credit = schedules.credit(week)	×
336	except schedulestorage.StorageError as e:	×
337	raise ApiError(e)	×
338	return yamlfy(**credit)	×
339
340	@app.get('/api/shifts/download/shiftload/{week}')	2✔
341	def downloadShiftLoad(week):	2✔
342	loadfile = Path('carrega-{}.csv'.format(week))	×
343
344	return FileResponse(	×
345	str(loadfile),
346	#as_attachment=True,
347	media_type='text/csv',
348	)
349
350	@app.get('/api/shifts/download/overload/{week}')	2✔
351	def downloadOverload(week):	2✔
352	loadfile = Path('overload-{}.yaml'.format(week))	×
353
354	return FileResponse(	×
355	str(loadfile),
356	#as_attachment=True,
357	media_type = 'application/x-yaml',
358	)
359
360	def yamlinfoerror(code, message, args, *kwds):	2✔
361	error(message, args, *kwds)	×
362	return yamlfy(info=ns(	×
363	info='',
364	message=code,
365	))
366
367	@app.get('/api/info/{field}/{value}')	2✔
368	def getInfoPersonBy(field, value):	2✔
369	decoded_field = urllib.parse.unquote(value)	×
370	data = None	×
371	with erp() as O:	×
372	callinfo = CallInfo(O)	×
373	try:	×
374	data = callinfo.getByField(field, decoded_field, shallow=True)	×
375	except ValueError:	×
376	return yamlinfoerror('error_getBy'+field.title(),	×
377	"Getting information searching {}='{}'.", field, value)
378	if data is None:	×
379	return 404	×
380	message = 'ok'	×
381	if not data.partners:	×
382	message = 'no_info'	×
383	elif data.partners == "Masses resultats":	×
384	message = 'response_too_long'	×
385
386	result = ns(	×
387	info=data,
388	message=message,
389	)
390	return yamlfy(info=result)	×
391
392	@app.post('/api/info/contractdetails')	2✔
393	async def getContractDetails(request: Request):	2✔
394	params = ns.loads(await request.body())	×
395	with erp() as O:	×
396	info = CallInfo(O)	×
397	data = info.contractDetails(params.contracts)	×
398
399	result = ns(	×
400	info=data,
401	message='ok',
402	)
403	return yamlfy(info=result)	×
404
405
406	@app.get('/api/info/ringring')	2✔
407	async def callingPhone(phone: str, extension: str):	2✔
408	return await notifyIncommingCall(phone, extension)	×
409
410	@app.post('/api/info/ringring')	2✔
411	async def callingPhonePost(phone: str = Form(...), ext: str = Form(...)):	2✔
412	return await notifyIncommingCall(phone, ext)	×
413
414	async def notifyIncommingCall(phone: str, extension: str):	2✔
415	time = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')	×
416	user = persons.byExtension(extension)	×
417	phone = phone.replace('+','').replace('-','').replace(' ','')	×
418	phone = re.sub(r'^[+]?0?0?34','', phone)	×
419	CallRegistry().updateCall(extension, fields=ns(	×
420	data = time,
421	telefon = phone,
422	motius = "",
423	partner = "",
424	contracte = "",
425	))
426	notifications = backchannel.notifyIncommingCall(user, phone, time)	×
427	if not notifications:	×
428	warn(	×
429	f"No sesion on extension {extension} "
430	f"to notify incomming call from {phone} at {time}")
431	await asyncio.gather(*notifications)	×
432	return yamlfy(result='ok')	×
433
434
435	@app.get('/api/personlog/{extension}')	2✔
436	def getCallLog(extension):	2✔
437	calls = CallRegistry().callsByExtension(extension)	×
438	return yamlfy(	×
439	info=ns(
440	info=calls,
441	message='ok',
442	)
443	)
444
445	@app.post('/api/call/annotate')	2✔
446	async def callAnnotate(request: Request):	2✔
NEW 447	annotation = ns.loads(await request.body())	×
NEW 448	CallRegistry().annotateCall(annotation)	×
NEW 449	user = annotation.get('user', None)	×
NEW 450	if user:	×
NEW 451	await asyncio.gather(	×
452	*backchannel.notifyCallLogChanged(user)
453	)
NEW 454	return yamlfy(info=ns(	×
455	message="ok"
456	))
457
458
459
460
461	@app.get('/api/updateClaims')	2✔
462	def updateClaimTypes():	2✔
463	with erp() as O:	×
464	CallRegistry().importClaimTypes(O)	×
465	return yamlfy(info=ns(message='ok'))	×
466
467
468	@app.get('/api/getClaims')	2✔
469	def getClaimTypes():	2✔
470	message = 'ok'	×
471	claims = CallRegistry().claimTypes()	×
472	if not claims:	×
473	error("File of claims does not exist")	×
474	return yamlfy(info=ns(	×
475	message="error",
476	claims=[],
477	dict={},
478	))
479	claims_dict = CallRegistry().claimKeywords()	×
480	if not claims_dict:	×
481	error("File of claims dict does not exist")	×
482	return yamlfy(info=ns(	×
483	message="error",
484	claims=[],
485	dict={},
486	))
487
488	result = ns(	×
489	message=message,
490	claims=claims,
491	dict=claims_dict,
492	)
493	return yamlfy(info=result)	×
494
495
496	@app.get('/api/updateCrmCategories')	2✔
497	def updateCrmCategories():	2✔
498	with erp() as O:	×
499	CallRegistry().importCrmCategories(O)	×
500	return yamlfy(info=ns(message='ok'))	×
501
502
503	@app.get('/api/getInfos')	2✔
504	def getInfos():	2✔
505	infos = CallRegistry().infoRequestTypes()	×
506	if not infos:	×
507	return yamlinfoerror("error",	×
508	"Unable to info request types")
509	return yamlfy(info = ns(	×
510	message='ok',
511	infos=infos,
512	))
513
514
515
516
517	# vim: ts=4 sw=4 et

Som-Energia / somenergia-tomatic / 1729921193

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous