Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
For now, just flush the entire calendar instead of being smart
about it. (This seems to be WebDAV's way of doing things
anyway)
"""
authorized = auth(web.ctx.env)
if not authorized:
auth_fail()
return
web.header('DAV', dav_header)
log('=== PUT ===')
# log('INPUT DATA')
# log(web.data())
ical = icalendar.Calendar.from_string(web.data())
ics = ical2cnode(ical)
erebus = ICS2ErebusVisitor(ics).run()
b = ExchangeBackend(host=host,https=False,user=username,
password=password)
old_items = b.get_all_item_ids()
ics = icalendar.Calendar.from_string(web.data())
cnode = ical2cnode(ics)
erebus = ICS2ErebusVisitor(cnode).run()
log(ToStringVisitor().visit(erebus))
b.create_item(erebus)
if old_items.search('event'):
b.delete_item(old_items)
def run_calendar_test(self):
self.test_calendar.set_mafia_calculator(self.datetester)
parsed_calendar=icalendar.Calendar.from_string(str(self.test_calendar))
curdate=datetime.date.today()
count=0
for component in parsed_calendar.walk():
if component.name=="VEVENT":
count+=1
curdate+=curdate.resolution
self.check_event(component,curdate)
self.assertEqual(count,self.daycount)
self.assertEqual(curdate,self.enddate)
def check_event(self,component,curdate):
def gather(self, url):
# Check when we last updated this URL
cal = Calendar.find().filter_by(url=url).first()
if cal and cal.ts and datetime.datetime.now() - cal.ts < datetime.timedelta(hours=1):
return
#pass
# Load the calendar
data = urllib2.urlopen(url).read()
data = data.replace('00001231T000000Z', '00011231T000000Z') # datetime doesn't like year=0
ical = icalendar.Calendar.from_string(data)
# Delete all existing events for this URL
CalendarEvent.find().filter_by(url=url).delete()
# Create the Calendar if needed
if not cal:
cal = Calendar(url=url).add()
cal.ts = datetime.datetime.now()
# Get the timezone for this calendar
now = datetime.datetime.now(tz.tzlocal())
for vtimezone in ical.walk('VTIMEZONE'):
sio = StringIO()
for line in str(vtimezone).splitlines():
if not line.lower().startswith('x-'):
sio.write(line)
def do_PUT(self):
"""Receive a todolist for updating"""
length = int(self.headers.getheader('content-length'))
cal = icalendar.Calendar.from_string(self.rfile.read(length))
for vTodo in cal.walk():
if "UID" in vTodo:
try:
self._processVTodo(vTodo)
except YokadiException, e:
self.send_response(503, e)
# Tell caller everything is ok
self.send_response(200)
self.end_headers()
self._log('Added uid mapping: %s -> %s' %(ical_uid, eid))
self.itemids[str(ical_uid)] = eid
self._log('Created item with id %s' % eid)
return 'Sucessfully uploaded calendar item'
elif self.handler.headers.has_key('If-Match'):
# Update this item
self._log('Updating item!')
etag = self.handler.headers['If-Match']
if self.itemids.has_key(etag):
ei = self.itemids[etag]
else:
ei = etag2exchange_id(etag)
ical = icalendar.Calendar.from_string(data)
ics = ical2cnode(ical)
item_changes = ICS2ErebusVisitor(ics).run()
b.update_item(ei, item_changes)
raise DAV_Error, 204
except QueryError, e:
if e.status == 401:
self.handler.send_autherror(401,"Authorization Required")
return
self._log(e)
raise
return DAV_Forbidden
def import_feed(self, feed):
from molly.apps.feeds.models import Item, vCard
calendar = Calendar.from_string(urllib2.urlopen(feed.rss_url).read())
items = set()
for component in calendar.walk():
if component.name == 'VEVENT':
item, created = Item.objects.get_or_create(feed=feed,
guid=str(component.get('UID')))
# Do not create the event if one the property is not correct,
# first tries to parse DT as datetime then as date, if it still
# fails, then ignore
try:
try:
item.dt_start = vDatetime.from_ical(str(
component.get('DTSTART')))
except ValueError, ve:
item.dt_start = vDate.from_ical(str(
def put(environ, service):
headers = []
resource = environ['PATH_INFO']
input = environ['wsgi.input']
data = input.read()
cal = icalendar.Calendar.from_string(data)
for event in cal.walk():
if event.name != 'VEVENT':
continue
if 'summary' in event:
summary = event['summary']
else:
summary = 'no summary'
tasks = service.tasks().list(tasklist=tasklist_id).execute()
for task in tasks['items']:
if task['title'] == summary:
return 'CONFLICT\n', httplib.CONFLICT, headers
task = {
'title': summary,
'notes': 'added from CalDav client',