@transactional_session
def __declare_bad_file_replicas(pfns, rse_id, reason, issuer, status=BadFilesStatus.BAD, scheme='srm', session=None):
    '\n    Declare a list of bad replicas.\n\n    :param pfns: The list of PFNs.\n    :param rse_id: The RSE id.\n    :param reason: The reason of the loss.\n    :param issuer: The issuer account.\n    :param status: Either BAD or SUSPICIOUS.\n    :param scheme: The scheme of the PFNs.\n    :param session: The database session in use.\n    '
    unknown_replicas = []
    declared_replicas = []
    rse_info = rsemgr.get_rse_info(rse_id=rse_id, session=session)
    replicas = []
    proto = rsemgr.create_protocol(rse_info, 'read', scheme=scheme)
    if rse_info['deterministic']:
        parsed_pfn = proto.parse_pfns(pfns=pfns)
        for pfn in parsed_pfn:
            path = parsed_pfn[pfn]['path']
            if (path.startswith('/user') or path.startswith('/group')):
                scope = ('%s.%s' % (path.split('/')[1], path.split('/')[2]))
                name = parsed_pfn[pfn]['name']
            elif path.startswith('/'):
                scope = path.split('/')[1]
                name = parsed_pfn[pfn]['name']
            else:
                scope = path.split('/')[0]
                name = parsed_pfn[pfn]['name']
            scope = InternalScope(scope, vo=issuer.vo)
            (__exists, scope, name, already_declared, size) = __exists_replicas(rse_id, scope, name, path=None, session=session)
            if (__exists and (((status == BadFilesStatus.BAD) and (not already_declared)) or (status == BadFilesStatus.SUSPICIOUS))):
                replicas.append({'scope': scope, 'name': name, 'rse_id': rse_id, 'state': ReplicaState.BAD})
                new_bad_replica = models.BadReplicas(scope=scope, name=name, rse_id=rse_id, reason=reason, state=status, account=issuer, bytes=size)
                new_bad_replica.save(session=session, flush=False)
                session.query(models.Source).filter_by(scope=scope, name=name, rse_id=rse_id).delete(synchronize_session=False)
                declared_replicas.append(pfn)
            elif already_declared:
                unknown_replicas.append(('%s %s' % (pfn, 'Already declared')))
            else:
                no_hidden_char = True
                for char in str(pfn):
                    if (not isprint(char)):
                        unknown_replicas.append(('%s %s' % (pfn, 'PFN contains hidden chars')))
                        no_hidden_char = False
                        break
                if no_hidden_char:
                    unknown_replicas.append(('%s %s' % (pfn, 'Unknown replica')))
        if (status == BadFilesStatus.BAD):
            try:
                update_replicas_states(replicas, session=session)
            except exception.UnsupportedOperation:
                raise exception.ReplicaNotFound("One or several replicas don't exist.")
    else:
        path_clause = []
        parsed_pfn = proto.parse_pfns(pfns=pfns)
        for pfn in parsed_pfn:
            path = ('%s%s' % (parsed_pfn[pfn]['path'], parsed_pfn[pfn]['name']))
            (__exists, scope, name, already_declared, size) = __exists_replicas(rse_id, scope=None, name=None, path=path, session=session)
            if (__exists and (((status == BadFilesStatus.BAD) and (not already_declared)) or (status == BadFilesStatus.SUSPICIOUS))):
                replicas.append({'scope': scope, 'name': name, 'rse_id': rse_id, 'state': ReplicaState.BAD})
                new_bad_replica = models.BadReplicas(scope=scope, name=name, rse_id=rse_id, reason=reason, state=status, account=issuer, bytes=size)
                new_bad_replica.save(session=session, flush=False)
                session.query(models.Source).filter_by(scope=scope, name=name, rse_id=rse_id).delete(synchronize_session=False)
                declared_replicas.append(pfn)
                path_clause.append((models.RSEFileAssociation.path == path))
                if path.startswith('/'):
                    path_clause.append((models.RSEFileAssociation.path == path[1:]))
                else:
                    path_clause.append((models.RSEFileAssociation.path == ('/%s' % path)))
            elif already_declared:
                unknown_replicas.append(('%s %s' % (pfn, 'Already declared')))
            else:
                no_hidden_char = True
                for char in str(pfn):
                    if (not isprint(char)):
                        unknown_replicas.append(('%s %s' % (pfn, 'PFN contains hidden chars')))
                        no_hidden_char = False
                        break
                if no_hidden_char:
                    unknown_replicas.append(('%s %s' % (pfn, 'Unknown replica')))
        if ((status == BadFilesStatus.BAD) and (declared_replicas != [])):
            query = session.query(models.RSEFileAssociation).with_hint(models.RSEFileAssociation, '+ index(replicas REPLICAS_PATH_IDX', 'oracle').filter((models.RSEFileAssociation.rse_id == rse_id)).filter(or_(*path_clause))
            rowcount = query.update({'state': ReplicaState.BAD})
            if (rowcount != len(declared_replicas)):
                print(rowcount, len(declared_replicas), declared_replicas)
                raise exception.ReplicaNotFound("One or several replicas don't exist.")
    try:
        session.flush()
    except IntegrityError as error:
        raise exception.RucioException(error.args)
    except DatabaseError as error:
        raise exception.RucioException(error.args)
    except FlushError as error:
        raise exception.RucioException(error.args)
    return unknown_replicas
Declare a list of bad replicas.

:param pfns: The list of PFNs.
:param rse_id: The RSE id.
:param reason: The reason of the loss.
:param issuer: The issuer account.
:param status: Either BAD or SUSPICIOUS.
:param scheme: The scheme of the PFNs.
:param session: The database session in use.
@transactional_session
def __declare_bad_file_replicas(pfns, rse_id, reason, issuer, status=BadFilesStatus.BAD, scheme='srm', session=None):
    '\n    Declare a list of bad replicas.\n\n    :param pfns: The list of PFNs.\n    :param rse_id: The RSE id.\n    :param reason: The reason of the loss.\n    :param issuer: The issuer account.\n    :param status: Either BAD or SUSPICIOUS.\n    :param scheme: The scheme of the PFNs.\n    :param session: The database session in use.\n    '
    unknown_replicas = []
    declared_replicas = []
    rse_info = rsemgr.get_rse_info(rse_id=rse_id, session=session)
    replicas = []
    proto = rsemgr.create_protocol(rse_info, 'read', scheme=scheme)
    if rse_info['deterministic']:
        parsed_pfn = proto.parse_pfns(pfns=pfns)
        for pfn in parsed_pfn:
            path = parsed_pfn[pfn]['path']
            if (path.startswith('/user') or path.startswith('/group')):
                scope = ('%s.%s' % (path.split('/')[1], path.split('/')[2]))
                name = parsed_pfn[pfn]['name']
            elif path.startswith('/'):
                scope = path.split('/')[1]
                name = parsed_pfn[pfn]['name']
            else:
                scope = path.split('/')[0]
                name = parsed_pfn[pfn]['name']
            scope = InternalScope(scope, vo=issuer.vo)
            (__exists, scope, name, already_declared, size) = __exists_replicas(rse_id, scope, name, path=None, session=session)
            if (__exists and (((status == BadFilesStatus.BAD) and (not already_declared)) or (status == BadFilesStatus.SUSPICIOUS))):
                replicas.append({'scope': scope, 'name': name, 'rse_id': rse_id, 'state': ReplicaState.BAD})
                new_bad_replica = models.BadReplicas(scope=scope, name=name, rse_id=rse_id, reason=reason, state=status, account=issuer, bytes=size)
                new_bad_replica.save(session=session, flush=False)
                session.query(models.Source).filter_by(scope=scope, name=name, rse_id=rse_id).delete(synchronize_session=False)
                declared_replicas.append(pfn)
            elif already_declared:
                unknown_replicas.append(('%s %s' % (pfn, 'Already declared')))
            else:
                no_hidden_char = True
                for char in str(pfn):
                    if (not isprint(char)):
                        unknown_replicas.append(('%s %s' % (pfn, 'PFN contains hidden chars')))
                        no_hidden_char = False
                        break
                if no_hidden_char:
                    unknown_replicas.append(('%s %s' % (pfn, 'Unknown replica')))
        if (status == BadFilesStatus.BAD):
            try:
                update_replicas_states(replicas, session=session)
            except exception.UnsupportedOperation:
                raise exception.ReplicaNotFound("One or several replicas don't exist.")
    else:
        path_clause = []
        parsed_pfn = proto.parse_pfns(pfns=pfns)
        for pfn in parsed_pfn:
            path = ('%s%s' % (parsed_pfn[pfn]['path'], parsed_pfn[pfn]['name']))
            (__exists, scope, name, already_declared, size) = __exists_replicas(rse_id, scope=None, name=None, path=path, session=session)
            if (__exists and (((status == BadFilesStatus.BAD) and (not already_declared)) or (status == BadFilesStatus.SUSPICIOUS))):
                replicas.append({'scope': scope, 'name': name, 'rse_id': rse_id, 'state': ReplicaState.BAD})
                new_bad_replica = models.BadReplicas(scope=scope, name=name, rse_id=rse_id, reason=reason, state=status, account=issuer, bytes=size)
                new_bad_replica.save(session=session, flush=False)
                session.query(models.Source).filter_by(scope=scope, name=name, rse_id=rse_id).delete(synchronize_session=False)
                declared_replicas.append(pfn)
                path_clause.append((models.RSEFileAssociation.path == path))
                if path.startswith('/'):
                    path_clause.append((models.RSEFileAssociation.path == path[1:]))
                else:
                    path_clause.append((models.RSEFileAssociation.path == ('/%s' % path)))
            elif already_declared:
                unknown_replicas.append(('%s %s' % (pfn, 'Already declared')))
            else:
                no_hidden_char = True
                for char in str(pfn):
                    if (not isprint(char)):
                        unknown_replicas.append(('%s %s' % (pfn, 'PFN contains hidden chars')))
                        no_hidden_char = False
                        break
                if no_hidden_char:
                    unknown_replicas.append(('%s %s' % (pfn, 'Unknown replica')))
        if ((status == BadFilesStatus.BAD) and (declared_replicas != [])):
            query = session.query(models.RSEFileAssociation).with_hint(models.RSEFileAssociation, '+ index(replicas REPLICAS_PATH_IDX', 'oracle').filter((models.RSEFileAssociation.rse_id == rse_id)).filter(or_(*path_clause))
            rowcount = query.update({'state': ReplicaState.BAD})
            if (rowcount != len(declared_replicas)):
                print(rowcount, len(declared_replicas), declared_replicas)
                raise exception.ReplicaNotFound("One or several replicas don't exist.")
    try:
        session.flush()
    except IntegrityError as error:
        raise exception.RucioException(error.args)
    except DatabaseError as error:
        raise exception.RucioException(error.args)
    except FlushError as error:
        raise exception.RucioException(error.args)
    return unknown_replicas