def is_valid_uri(references):
    return bool(URI.match(references)) PW_AGENT = URIRef("http://data.wu.ac.at/portalwatch") class DBManager(object):
    def __init__(self, db='portalwatch', host="localhost", port=5432, password=None, user='opwu', debug=False):

        # Define our connection string
        self.log = log.new()

        conn_string = "postgresql://"
        if user:
            conn_string += user
        if password:
            conn_string += ":" + password
        if host:
            conn_string += "@" + host
        if port:
            conn_string += ":" + str(port)
        conn_string += "/" + db
        log.info("Connecting DB")

        self.engine = create_engine(conn_string, pool_size=20, client_encoding='utf8', echo=debug)
        add_engine_pidguard(self.engine)
        #add_query_time_logging(self.engine)
        # register_after_fork(self.engine, self.engine.dispose)
        log.info("Connected DB")
        # self.engine.connect()


        self.session_factory = sessionmaker(bind=self.engine)  # , expire_on_commit=False

        # self.session = self.Session()



        self.dataset_insert_function = DDL(
            """
            CREATE OR REPLACE FUNCTION dataset_insert_function()
            RETURNS TRIGGER AS $$

            DECLARE
                _snapshot smallint;
                _table_name text;

            BEGIN
                _snapshot := NEW.snapshot;
                _table_name := '""" + tab_datasets + """_' || _snapshot;

                    PERFORM 1 FROM pg_tables WHERE tablename = _table_name;

                      IF NOT FOUND THEN
                        EXECUTE
                          'CREATE TABLE '
                          || quote_ident(_table_name)
                          || ' (CHECK ("snapshot" = '
                          || _snapshot::smallint
                          || ')) INHERITS (""" + tab_datasets + """)';

                        -- Indexes are defined per child, so we assign a default index that uses the partition columns
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_org') || ' ON '||quote_ident(_table_name) || ' (organisation)';
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_sn')  || ' ON '||quote_ident(_table_name) || ' (snapshot)';
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_pid') || ' ON '||quote_ident(_table_name) || ' (portalid)';
                        EXECUTE 'ALTER TABLE '  || quote_ident(_table_name)|| ' ADD CONSTRAINT ' || quote_ident(_table_name||'_pkey') || ' PRIMARY KEY (id, snapshot, portalid)';
                        EXECUTE 'ALTER TABLE '  || quote_ident(_table_name)|| ' ADD CONSTRAINT ' || quote_ident(_table_name||'_md5_fkey') || ' FOREIGN KEY (md5) REFERENCES datasetsdata (md5) MATCH SIMPLE ON UPDATE NO ACTION ON DELETE NO ACTION';
                        EXECUTE 'ALTER TABLE '  || quote_ident(_table_name)|| ' ADD CONSTRAINT ' || quote_ident(_table_name||'_portalid_fkey') || ' FOREIGN KEY (portalid, snapshot) REFERENCES portalsnapshot (portalid, snapshot) MATCH SIMPLE ON UPDATE NO ACTION ON DELETE NO ACTION';


                      END IF;

                      EXECUTE
                        'INSERT INTO '
                        || quote_ident(_table_name)
                        || ' VALUES ($1.*)'
                      USING NEW;

                      RETURN NULL;
                END;

                $$ LANGUAGE plpgsql;
                """)
        self.dataset_insert_trigger = DDL(
            """
            CREATE TRIGGER dataset_insert_trigger
            BEFORE INSERT ON """ + tab_datasets + """
                FOR EACH ROW EXECUTE PROCEDURE dataset_insert_function();
                """
        )
        self.resourcesinfo_insert_function = DDL(
            """
            CREATE OR REPLACE FUNCTION resourcesinfo_insert_function()
            RETURNS TRIGGER AS $$

            DECLARE
                _snapshot smallint;
                _table_name text;

            BEGIN
                _snapshot := NEW.snapshot;
                _table_name := '""" + tab_resourcesinfo + """_' || _snapshot;

                    PERFORM 1 FROM pg_tables WHERE tablename = _table_name;

                      IF NOT FOUND THEN
                        EXECUTE
                          'CREATE TABLE '
                          || quote_ident(_table_name)
                          || ' (CHECK ("snapshot" = '
                          || _snapshot::smallint
                          || ')) INHERITS (""" + tab_resourcesinfo + """)';

                        -- Indexes are defined per child, so we assign a default index that uses the partition columns
                        EXECUTE 'ALTER TABLE '  || quote_ident(_table_name)|| ' ADD CONSTRAINT ' || quote_ident(_table_name||'_pkey') || ' PRIMARY KEY (uri, snapshot)';
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_status') || ' ON '||quote_ident(_table_name) || ' (status)';
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_uri') || ' ON '||quote_ident(_table_name) || ' (uri)';
                      END IF;

                      EXECUTE
                        'INSERT INTO '
                        || quote_ident(_table_name)
                        || ' VALUES ($1.*)'
                      USING NEW;



                      RETURN NULL;
                END;

                $$ LANGUAGE plpgsql;
                """)
        self.resourcesinfo_insert_trigger = DDL(
            """
            CREATE TRIGGER resourcesinfo_insert_trigger
            BEFORE INSERT ON """ + tab_resourcesinfo + """
                FOR EACH ROW EXECUTE PROCEDURE resourcesinfo_insert_function();
                """
        )

        self.resourcescrawllog_insert_function = DDL(
            """
            CREATE OR REPLACE FUNCTION resourcescrawllog_insert_function()
            RETURNS TRIGGER AS $$

            DECLARE
                _snapshot smallint;
                _table_name text;

            BEGIN
                _snapshot := NEW.snapshot;
                _table_name := '""" + tab_resourcescrawllog + """_' || _snapshot;

                    PERFORM 1 FROM pg_tables WHERE tablename = _table_name;

                      IF NOT FOUND THEN
                        EXECUTE
                          'CREATE TABLE '
                          || quote_ident(_table_name)
                          || ' (CHECK ("snapshot" = '
                          || _snapshot::smallint
                          || ')) INHERITS (""" + tab_resourcescrawllog + """)';

                        -- Indexes are defined per child, so we assign a default index that uses the partition columns
                        EXECUTE 'ALTER TABLE '  || quote_ident(_table_name)|| ' ADD CONSTRAINT ' || quote_ident(_table_name||'_pkey') || ' PRIMARY KEY (uri, snapshot, timestamp)';
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_status') || ' ON '||quote_ident(_table_name) || ' (status)';
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_uri') || ' ON '||quote_ident(_table_name) || ' (uri)';
                        EXECUTE 'CREATE INDEX ' || quote_ident(_table_name||'_domain') || ' ON '||quote_ident(_table_name) || ' (domain)';
                      END IF;

                      EXECUTE
                        'INSERT INTO '
                        || quote_ident(_table_name)
                        || ' VALUES ($1.*)'
                      USING NEW;



                      RETURN NULL;
                END;

                $$ LANGUAGE plpgsql;
                """)
        self.resourcescrawllog_insert_trigger = DDL(
            """
            CREATE TRIGGER resourcescrawllog_insert_trigger
            BEFORE INSERT ON """ + tab_resourcescrawllog + """
                FOR EACH ROW EXECUTE PROCEDURE resourcescrawllog_insert_function();
                """
        )

    def db_DropEverything(self):
        print "Sroping everything"
        conn = self.engine.connect()

        # the transaction only applies if the DB supports
        # transactional DDL, i.e. Postgresql, MS SQL Server
        trans = conn.begin()

        inspector = reflection.Inspector.from_engine(self.engine)

        # gather all data first before dropping anything.
        # some DBs lock after things have been dropped in
        # a transaction.

        metadata = MetaData()

        tbs = []
        all_fks = []

        for table_name in inspector.get_table_names():
            fks = []
            for fk in inspector.get_foreign_keys(table_name):
                if not fk['name']:
                    continue
                fks.append(
                    ForeignKeyConstraint((), (), name=fk['name'])
                )
            t = Table(table_name, metadata, *fks)
            tbs.append(t)
            all_fks.extend(fks)

        for fkc in all_fks:
            conn.execute(DropConstraint(fkc, cascade=True))

        for table in tbs:
            if table.name != tab_datasets and table.name != tab_resourcesinfo:
                conn.execute(DropTable(table))

        trans.commit()

    def tableExists(self,Variable_tableName):
        return self.engine.has_table(self.engine, Variable_tableName)
    def viewExists(self,Variable_tableName):
        print 'check if table {} exists'.format(Variable_tableName)
        return self.engine.connect().execute()

    def init(self, Base):

        event.listen(Dataset.__table__, 'after_create', self.dataset_insert_function)
        event.listen(Dataset.__table__, 'after_create', self.dataset_insert_trigger)
        event.listen(ResourceInfo.__table__, 'after_create', self.resourcesinfo_insert_function)
        event.listen(ResourceInfo.__table__, 'after_create', self.resourcesinfo_insert_trigger)

        Base.metadata.create_all(self.engine) def profile(table):
def isfloat(value):
def isdatetime(datestring):
def get_label(v):
    def __init__(self, values):
    def _classify(self):
class ColumnLabel:
class Column:
    NUMBER = XSD.number
    ANY = XSD.anyAtomicType
    BINARY = XSD.binary
    DATETIME = XSD.datetime
    EMPTY = XSD.anyAtomicType class DBClient(object):

    def __init__(self, dbm=None, Session=None):
        if dbm is not None:
            self.dbm=dbm
            self.Session = scoped_session(dbm.session_factory)
        elif Session:
            self.Session
        Base.query = self.Session.query_property()


    from contextlib import contextmanager

    @contextmanager
    def session_scope(self):
        """Provide a transactional scope around a series of operations."""
        #session = self.Session()
        try:
            yield self.Session
            self.Session.flush()
            self.Session.commit()
        except:
            self.Session.rollback()
            raise
        #finally:
        #   self.Session.remove()

    def remove(self):
        self.Session.remove()

    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#
    ### ADD & COMMIT
    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#

    def delete(self, obj):
        with self.session_scope() as session:
            session.delete(obj)

    def add(self, obj):
        with self.session_scope() as session:
            session.add(obj)

    def bulkadd(self, obj):
        with self.session_scope() as session:
            session.bulk_save_objects(obj)

    def commit(self):
        self.Session.commit()

    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#
    ### EXISTS FUNCTIONS
    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#

    def exist_datasetdata(self, md5):
        return DatasetData.query.filter_by(md5=md5).first()

    def exist_resourceinfo(self, uri, snapshot):
        return ResourceInfo.query.filter_by(uri=uri, snapshot=snapshot).first()

    def exist_metaresource(self, uri):
        return MetaResource.query.filter_by(uri=uri).first()

    def datasetqualityExists(self, md5):
        with self.session_scope() as session:
            return session.query(DatasetQuality).filter_by(md5=md5).first()

    def getContentLocation(self, uri=None, digest=None):
        with self.session_scope() as session:
            q= session.query(ResourceCrawlLog.disklocation).filter(ResourceCrawlLog.uri==uri).filter(ResourceCrawlLog.digest==digest).first()
            return q
    def getLastDigest(self, uri):
        with self.session_scope() as session:
            q= session.query(ResourceCrawlLog.digest).filter(ResourceCrawlLog.uri==uri).filter(ResourceCrawlLog.status==200).order_by(ResourceCrawlLog.timestamp.desc()).first()
            return q

    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#
    ### PORTALS
    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#
    def portals(self):
        with self.session_scope() as session:
            return session.query(Portal)

    def portalAll(self):
        with self.session_scope() as session:
            return session.query(Portal, Portal.snapshot_count,Portal.first_snapshot, Portal.last_snapshot, Portal.datasetCount, Portal.resourceCount)

    def portalsSnapshots(self, snapshot):
        with self.session_scope() as session:
            return session.query(PortalSnapshot)\
            .filter(PortalSnapshot.snapshot==snapshot)


    def portalsQuality(self,snapshot):
        with self.session_scope() as session:
            return session.query(PortalSnapshotQuality)\
            .filter(PortalSnapshotQuality.snapshot==snapshot)

    def portalsAll(self,snapshot):
        with self.session_scope() as session:
            return session.query( PortalSnapshot)\
                .filter(PortalSnapshot.snapshot==snapshot)\
                .outerjoin( PortalSnapshotQuality, PortalSnapshot.portalid==PortalSnapshotQuality.portalid )\
                .join(Portal)\
                .add_entity(PortalSnapshotQuality)\
                .add_entity(Portal)\


    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#
    ### PORTAL
    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#

    def portal(self, portalid):
        with self.session_scope() as session:
            return session.query(Portal).filter(Portal.id==portalid)

    def portalSnapshot(self,snapshot, portalid):
        with self.session_scope() as session:
            return session.query(PortalSnapshot)\
                .filter(PortalSnapshot.snapshot==snapshot)\
                .filter(PortalSnapshot.portalid==portalid)

    def portalSnapshotQuality(self, portalid, snapshot):
        with self.session_scope() as session:
            return session.query(PortalSnapshotQuality)\
                .filter(PortalSnapshotQuality.portalid==portalid)\
                .filter(PortalSnapshotQuality.snapshot==snapshot)


    def statusCodeDist(self, snapshot,portalid=None):
        with self.session_scope() as session:
            q= session.query(ResourceInfo.status,func.count())\
                .join(MetaResource, ResourceInfo.uri==MetaResource.uri)\
                .join(Dataset,Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)\
                .filter(ResourceInfo.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)

            q=q.group_by(ResourceInfo.status)
            q=q.order_by(func.count().desc())

            return q

    def organisationDist(self, snapshot, portalid=None):
        with self.session_scope() as session:
            q= session.query(Dataset.organisation,func.count().label('count'))\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            q=q.group_by(Dataset.organisation)
            q=q.order_by(func.count().desc())
            return q



    def formatDist(self, snapshot, portalid=None, iso=None, software=None):
        with self.session_scope() as session:
            q= session.query(MetaResource.format, func.count().label('count'))\
                .join(Dataset,Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            if iso:
                q= q.join(Portal, Portal.id == Dataset.portalid)\
                .filter(Portal.iso==iso)
            if software:
                q= q.join(Portal, Portal.id == Dataset.portalid)\
                .filter(Portal.software==software)


            q=q.group_by(MetaResource.format)
            q=q.order_by(func.count().desc())
            return q


    def distinctFormats(self, snapshot, portalid=None):
        with self.session_scope() as session:
            q= session.query(MetaResource).distinct(MetaResource.format)\
                .join(Dataset,Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            return q

    def licenseDist(self, snapshot, portalid=None):
        with self.session_scope() as session:
            q= session.query(DatasetData.license,func.count().label('count')).join(Dataset)\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            q=q.group_by(DatasetData.license)
            q=q.order_by(func.count().desc())
            return q

    def distinctLicenses(self, snapshot, portalid=None):
        with self.session_scope() as session:
            q= session.query(DatasetData).distinct(DatasetData.license)\
                .join(Dataset)\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            return q

    def validURLDist(self, snapshot,portalid=None):
        with self.session_scope() as session:
            q= session.query(MetaResource.valid, func.count().label('count'))\
                .join(Dataset,Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)

            q=q.group_by(MetaResource.valid)
            q=q.order_by(func.count(MetaResource.valid).desc())

            return q

    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#
    ### DATASETS
    #--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--*--#

    def getDatasets(self, portalid=None, snapshot=None):
        with self.session_scope() as session:
            q= session.query(Dataset)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            if snapshot:
                q=q.filter(Dataset.snapshot==snapshot)
            return q

    def getDatasetData(self,md5=None):
        with self.session_scope() as session:
            q= session.query(DatasetData).filter(DatasetData.md5==md5).first()
            return q


    def getUnfetchedResources(self,snapshot, portalid=None, batch=None, iso=None, exclude_iso=None):
        with self.session_scope() as session:
            q=session.query(MetaResource.uri)\
                .join(Dataset, Dataset.md5==MetaResource.md5)\
                .join(Portal, Portal.id==Dataset.portalid)\
                .filter(Dataset.snapshot==snapshot)\
                .filter(MetaResource.valid==True)\
                .filter(
                #~exists().where(ResourceInfo.uri == MetaResource.uri).where(ResourceInfo.snapshot == snapshot)
                    ~exists().where(
                        and_( ResourceInfo.uri==MetaResource.uri, ResourceInfo.snapshot==snapshot))
                )
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            if iso:
                q=q.filter(Portal.iso==iso)
            if exclude_iso:
                q=q.filter(Portal.iso!=iso)
            if batch:
                q=q.limit(batch)
            return q

    def getDataUnfetchedResources(self,snapshot, portalid=None, batch=None, format=None):
        with self.session_scope() as session:
            q=session.query(MetaResource.uri, Dataset.id, Dataset.md5)\
                .join(Dataset, Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)\
                .filter(MetaResource.valid==True)\
                .filter(
                    ~exists().where(and_(MetaResource.uri== ResourceCrawlLog.uri, ResourceCrawlLog.snapshot==snapshot))
                )
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            if batch:
                q=q.limit(batch)
            if format:
                q=q.filter(MetaResource.format==format)
            return q

    def getDataResources(self, snapshot, portalid=None, format=None):
        with self.session_scope() as session:
            q=session.query(MetaResource.uri, Dataset.id, Dataset.md5)\
                .join(Dataset, Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)\
                .filter(MetaResource.valid==True)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            if format:
                q=q.filter(MetaResource.format==format)
            return q

    def getResourceInfos(self, snapshot, portalid=None):
        with self.session_scope() as session:
            q= session.query(ResourceInfo)\
                .join(MetaResource, ResourceInfo.uri==MetaResource.uri)\
                .join(Dataset,Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            return q


    def getMetaResource(self, snapshot, portalid=None, size=None):
        with self.session_scope() as session:
            q= session.query(MetaResource)\
                .join(Dataset, Dataset.md5==MetaResource.md5)\
                .filter(Dataset.snapshot==snapshot)
            if portalid:
                q=q.filter(Dataset.portalid==portalid)
            if size:
                q=q.filter((MetaResource.size<=size) | (MetaResource.size == None))
            return q

    def getResourceInfoByURI(self, uri, snapshot):
        with self.session_scope() as session:
            q= session.query(ResourceInfo)\
                .filter(ResourceInfo.uri==uri)\
                .filter(ResourceInfo.snapshot==snapshot)
            return q

    def portalSnapshotQualityDF(self, portalid, snapshot):

        q= self.portalSnapshotQuality(portalid,snapshot)
        data=None
        for r in q:
            data=row2dict(r)
            break
        d=[]

        datasets= int(data['datasets'])
        for inD in qa:
            for k , v in inD['metrics'].items():
                k=k.lower()
                # TODO what to do if metric has no value?
                if data[k] != None and data[k] != 'None':
                    value=float(data[k])
                    perc=int(data[k+'N'])/(datasets*1.0) if datasets>0 else 0
                    c= { 'Metric':k, 'Dimension':inD['dimension'],
                         'dim_color':inD['color'], 'value':value, 'perc':perc}
                    c.update(v)
                    d.append(c)
        return pd.DataFrame(d)

    def getResourcesHistory(self, uri, md5, source=None):
        with self.session_scope() as session:
            q= session.query(ResourceHistory)\
                .filter(ResourceHistory.uri==uri)\
                .filter(ResourceHistory.md5==md5)
            if source:
                q=q.filter(ResourceHistory.source==source)
            q=q.order_by(ResourceHistory.snapshot.asc())
            return q

    def getPortalSnapshotDynamics(self, snapshot, portalid):
        with self.session_scope() as session:
            q = session.query(PortalSnapshotDynamicity)\
                .filter(PortalSnapshotDynamicity.snapshot==snapshot)\
                .filter(PortalSnapshotDynamicity.portalid==portalid)
            return q def readDBConfFromFile(config):
    dbConf={ 'host':None, 'port':None, 'password':None, 'user':None, 'db':None}
    if config:
        with open(config) as f_conf:
            config = yaml.load(f_conf)
            if config is not None and 'db' in config:
                dbConf.update(config['db'])
            if config is None:
                log.error("Cannot load config file")
    #log.info("DBConfig", conf=dbConf)
    return dbConf def getCurrentSnapshot():
    now = datetime.now()
    y=now.isocalendar()[0]
    w=now.isocalendar()[1]
    sn=str(y)[2:]+'{:02}'.format(w)

    return sn def tofirstdayinisoweek(yearweek):
    """

    :param yearweek:
    :return: the first day in a week
    """
    year=int('20'+str(yearweek)[:2])
    week=int(str(yearweek)[2:])

    d = date(year,1,1)
    d = d - timedelta(d.weekday())
    dlt = timedelta(days = (week)*7)
    return d + dlt def toLastdayinisoweek(yearweek):

    year=int('20'+str(yearweek)[:2])
    week=int(str(yearweek)[2:])
    d = date(year,1,1)
    d = d - timedelta(d.weekday())
    dlt = timedelta(days = (week)*7)
    return d + dlt + timedelta(days=6) class Dataset(Base):
    __tablename__ = tab_datasets

    id           = Column( String, primary_key=True)
    snapshot     = Column( SmallInteger, primary_key=True, index=True)
    portalid     = Column( String, primary_key=True, index=True)
    organisation = Column(String, index=True)
    title        = Column(String, index=True)
    md5          = Column(String, ForeignKey(tab_datasetsdata+'.md5'), index=True)


    __table_args__ = (ForeignKeyConstraint([portalid, snapshot],
                                           [tab_portalsnapshot+'.portalid',tab_portalsnapshot+'.snapshot']),
                      {})

    portalsnapshot  = relationship("PortalSnapshot", back_populates="datasets")
    data  = relationship("DatasetData", back_populates="dataset")

    def __repr__(self):
        return "<Dataset(id=%s, portalid='%s', snapshot=%s, md5=%s)>" % (
            self.id, self.portalid, self.snapshot, self.md5) class MetaResource(Base):
    __tablename__ = tab_resources

    uri = Column(String, primary_key=True, index=True)
    md5 = Column(String,ForeignKey(DatasetData.md5), primary_key=True,index=True )
    valid = Column(Boolean)
    format = Column(String)
    media = Column(String)
    size = Column(BigInteger)
    created = Column(TIMESTAMP)
    modified = Column(TIMESTAMP)


    dataset  = relationship("DatasetData", back_populates="resources")


    def info(cls):
        return select([ResourceInfo]).where(cls.uri == ResourceInfo.uri).where(cls.snapshot== ResourceInfo.snapshot)

    def __repr__(self):
        return "<Resource(uri=%s, dataset=%s)>" % (
            self.uri, self.md5) class Portal(Base):
    __tablename__ = tab_portals

    id      = Column(String, primary_key=True, index=True,nullable=False)
    uri     = Column(String, nullable=False)
    apiuri  = Column(String)
    software = Column(String(12), nullable=False) # OpenDataSoft, CKAN, Socrata <13
    iso     = Column(String(2), nullable=False)
    active  = Column(Boolean, default=True,nullable=False)

    snapshots = relationship("PortalSnapshot", back_populates="portal")
    snapshotsquality = relationship("PortalSnapshotQuality", back_populates="portal")

    @hybrid_property
    def snapshot_count(self):
        print len(self.snapshots)
        return len(self.snapshots)
    @snapshot_count.expression

    def snapshot_count(cls):
        return select([func.count(PortalSnapshot.snapshot)])\
            .where(PortalSnapshot.portalid == cls.id).label("snapshot_count")

    @hybrid_property
    def first_snapshot(self):
        print [s for s in self.snapshots]
        return min([s.snapshot for s in self.snapshots])

    @first_snapshot.expression
    def first_snapshot(cls):
        return select([func.min(PortalSnapshot.snapshot)])\
            .where(PortalSnapshot.portalid == cls.id).label("first_snapshot")

    @hybrid_property
    def last_snapshot(self):
        return max([s.snapshot for s in self.snapshots])

    @last_snapshot.expression
    def last_snapshot(cls):
        return select([func.max(PortalSnapshot.snapshot)])\
            .where(PortalSnapshot.portalid == cls.id).label("last_snapshot")

    @hybrid_property
    def datasetcount(self):
        return self.snapshots.order_by(PortalSnapshot.snapshot.desc()).one().datasetcount

    @datasetcount.expression
    def datasetcount(cls):
        q=select([PortalSnapshot.datasetcount])\
            .where(PortalSnapshot.portalid == cls.id).order_by(PortalSnapshot.snapshot.desc()).limit(1).label("datasetcount")
        return q

    @hybrid_property
    def resourcecount(self):
        return self.snapshots.order_by(PortalSnapshot.snapshot.desc()).one().resourcecount

    @resourcecount.expression
    def resourcecount(cls):
        q=select([PortalSnapshot.resourcecount])\
            .where(PortalSnapshot.portalid == cls.id).order_by(PortalSnapshot.snapshot.desc()).limit(1).label("resourcecount")
        return q

    def __repr__(self):
        return "<Portal(id=%s, uri='%s', apiuri='%s', software='%s', iso=%s)>" % (
            self.id, self.uri, self.apiuri, self.software, self.iso)
import os
import pprint
import datetime
import requests
import structlog
import rdflib
import hashlib
from collections import defaultdict
from multiprocessing import Pool
from pyyacp import yacp
from rdflib import URIRef, BNode, Literal
from rdflib.namespace import Namespace, RDF, RDFS, DCTERMS, XSD
from odpw.core.dataset_converter import is_valid_uri
from odpw.quality.dqv_export import PW_AGENT
from odpw.core.db import DBManager
from odpw.csvanalyser import profiler
from odpw.core.api import DBClient
from odpw.utils.helper_functions import readDBConfFromFile
from odpw.utils.utils_snapshot import getCurrentSnapshot, tofirstdayinisoweek, toLastdayinisoweek
from odpw.core.model import Dataset, MetaResource, Portal


def addMetadata(url, snapshot, graph, max_lines=100, csvw_activity=None):
    timer = datetime.datetime.now()
    r = requests.get(url, stream=True, timeout=5)
    # BNode: url + snapshot
    bnode_hash = hashlib.sha1(url + str(snapshot))
    resource = BNode(bnode_hash.hexdigest())

    if is_valid_uri(url):
        ref = URIRef(url)
    else:
        ref = Literal(url)

    status_code = r.status_code
    if status_code == 200:
        # read first lines
        content = ''
        for i, line in enumerate(r.iter_lines()):
            content += line + '\n'
            if i >= max_lines:
                break
            if datetime.datetime.now() - timer > datetime.timedelta(seconds=3):
                break

        # initially test if we can parse the csv. if we fail, we do not create an entry
        parser = yacp.YACParser(content=content)

        # add url to graph
        graph.add((resource, CSVW.url, ref))

        if csvw_activity:
            graph.add((resource, PROV.wasGeneratedBy, csvw_activity))
            graph.add((csvw_activity, PROV.generated, resource))
            graph.add((resource, RDF.type, PROV.Entity))

        # add content type header information
        content_type = None
        if 'content-type' in r.headers:
            content_type = r.headers['content-type']
        elif 'Content-Type' in r.headers:
            content_type = r.headers['Content-Type']
        if content_type:
            try:
                # first part of content type
                t = content_type.split(';')[0]
                graph.add((resource, DCAT.mediaType, Literal(t)))
            except Exception as e:
                pass

        # add content length header information
        content_length = None
        if 'content-length' in r.headers:
            content_length = r.headers['content-length']
        elif 'Content-Length' in r.headers:
            content_length = r.headers['Content-Length']
        if content_length:
            try:
                # try to convert to numeric value
                l = float(content_length)
                graph.add((resource, DCAT.byteSize, Literal(l)))
            except Exception as e:
                pass

        data = profiler.profile(parser)

        # dialect
        # BNode: url + snapshot + CSVW.dialect
        bnode_hash = hashlib.sha1(url + str(snapshot) + CSVW.dialect.n3())
        dialect = BNode(bnode_hash.hexdigest())
        graph.add((resource, CSVW.dialect, dialect))
        graph.add((dialect, CSVW.encoding, Literal(data['encoding'])))
        graph.add((dialect, CSVW.delimiter, Literal(data['delimiter'])))
        if parser.description > 0:
            graph.add((dialect, CSVW.skipBlankRows, Literal(parser.description)))
        if data['header']:
            graph.add((dialect, CSVW.header, Literal(True)))
            graph.add((dialect, CSVW.headerRowCount, Literal(1)))
        else:
            graph.add((dialect, CSVW.header, Literal(False)))
            graph.add((dialect, CSVW.headerRowCount, Literal(0)))
        # columns
        # BNode: url + snapshot + CSVW.tableSchema
        bnode_hash = hashlib.sha1(url + str(snapshot) + CSVW.tableSchema.n3())
        tableschema = BNode(bnode_hash.hexdigest())
        graph.add((resource, CSVW.tableSchema, tableschema))
        col_i = 0
        for h, t in zip(data['header'], data['types']):
            # BNode: url + snapshot + CSVW.column + col_i
            bnode_hash = hashlib.sha1(url + str(snapshot) + CSVW.column.n3() + str(col_i))
            column = BNode(bnode_hash)
            graph.add((tableschema, CSVW.column, column))
            graph.add((column, CSVW.name, Literal(h)))
            graph.add((column, CSVW.datatype, t))
            col_i += 1
    # close request
    r.close()
    return status_code


def storeGraph(graph, portalid, directory):
    destination=os.path.join(directory, portalid + '.n3')
    graph.serialize(destination=destination, format='n3')
    log.info("CSVW Metadata graph stored", portal=portalid, destination=destination)


def csvw_prov(graph, snapshot):
    csvw_activity = URIRef("http://data.wu.ac.at/portalwatch/csvw/" + str(snapshot))
    graph.add((csvw_activity, RDF.type, PROV.Activity))
    graph.add((csvw_activity, PROV.startedAtTime, Literal(tofirstdayinisoweek(snapshot))))
    graph.add((csvw_activity, PROV.endedAtTime, Literal(toLastdayinisoweek(snapshot))))
    graph.add((csvw_activity, PROV.wasAssociatedWith, PW_AGENT))
    return csvw_activity


def streamCSVs(obj):
    P, dbConf, snapshot, dir = obj[0],obj[1],obj[2],obj[3]
    log.info("streamCSVs", portalid=P.id, snapshot=snapshot)

    dbm = DBManager(**dbConf)

    db = DBClient(dbm)